+service.consumer-base;

liquidnet-bus-common/liquidnet-common-service-base/src/main/java/com/liquidnet/service/base/MdbMessage.java

+package com.liquidnet.service.base;
+
+import com.liquidnet.commons.lang.util.JsonUtils;
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class MdbMessage implements Serializable, Cloneable {
+    private static final long serialVersionUID = 4814549190217810630L;
+    /**
+     * Mongo集合名
+     */
+    private String collect;
+    /**
+     * Mongo集合中的字段属性名称
+     */
+    private String column;
+    /**
+     * Mongo集合中的字段属性值
+     */
+    private String bizId;
+    /**
+     * Redis缓存中的Key前缀
+     */
+    private String prefix;
+    /**
+     * 操作类型[1-insert|2-update]
+     */
+    private int opType;
+
+    private static final MdbMessage obj = new MdbMessage();
+
+    public static MdbMessage getNew() {
+        try {
+            return (MdbMessage) obj.clone();
+        } catch (CloneNotSupportedException e) {
+            return new MdbMessage();
+        }
+    }
+
+    public String toJson() {
+        return JsonUtils.toJson(this);
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>liquidnet-service-consumer-all</artifactId>
+        <groupId>com.liquidnet</groupId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>liquidnet-service-consumer-base</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.liquidnet</groupId>
+            <artifactId>liquidnet-common-cache-redis</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
+
+        <!-- liquidnet-bus-api -->
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <filtering>true</filtering>
+            </resource>
+        </resources>
+    </build>
+</project>
\ No newline at end of file

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/ServiceConsumerBaseApplication.java

+package com.liquidnet.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.core.env.Environment;
+
+import java.net.InetAddress;
+import java.util.Arrays;
+
+@Slf4j
+@SpringBootApplication(scanBasePackages = {"com.liquidnet"})
+public class ServiceConsumerBaseApplication implements CommandLineRunner {
+    @Autowired
+    private Environment environment;
+
+    public static void main(String[] args) {
+        SpringApplication.run(ServiceConsumerBaseApplication.class, args);
+    }
+
+    @Override
+    public void run(String... strings) {
+        try {
+            log.info("\n----------------------------------------------------------\n\t" +
+                            "Application '{}' is running! Access URLs:\n\t" +
+                            "Local: \t\thttp://127.0.0.1:{}\n\t" +
+                            "External: \thttp://{}:{}{}/doc.html\n\t" +
+                            "Profile(s): \t{}\n----------------------------------------------------------",
+                    environment.getProperty("spring.application.name"),
+                    environment.getProperty("server.port"),
+                    InetAddress.getLocalHost().getHostAddress(),
+                    environment.getProperty("server.port"),
+                    environment.getProperty("server.servlet.context-path"),
+                    Arrays.toString(environment.getActiveProfiles()));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/config/ConsumerCommonSqlRedisStreamConfig.java

+package com.liquidnet.service.consumer.base.config;
+
+import com.liquidnet.common.cache.redis.config.RedisStreamConfig;
+import com.liquidnet.service.base.constant.MQConst;
+import com.liquidnet.service.consumer.base.receiver.ConsumerCommonSQL0Receiver;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.stream.Consumer;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.ReadOffset;
+import org.springframework.data.redis.connection.stream.StreamOffset;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamMessageListenerContainer;
+import org.springframework.data.redis.stream.Subscription;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 公共的SQL队列消息消费器初始化配置
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2022/6/2
+ */
+@Configuration
+public class ConsumerCommonSqlRedisStreamConfig extends RedisStreamConfig {
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+    @Autowired
+    ConsumerCommonSQL0Receiver consumerCommonSQL0Receiver;
+
+    @Bean
+    public List<Subscription> subscriptionSQL0(RedisConnectionFactory factory) {
+        List<Subscription> subscriptionList = new ArrayList<>();
+        MQConst.CommonQueue stream = MQConst.CommonQueue.SQL0;
+        this.initStream(stringRedisTemplate, stream.getKey(), stream.getGroup());
+        for (int i = 0; i < 5; i++) {
+            StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = this.buildStreamMessageListenerContainer(factory);
+            subscriptionList.add(listenerContainer.receiveAutoAck(
+                    Consumer.from(stream.getGroup(), getConsumerName(stream.name() + i)),
+                    StreamOffset.create(stream.getKey(), ReadOffset.lastConsumed()), consumerCommonSQL0Receiver
+            ));
+            listenerContainer.start();
+        }
+        return subscriptionList;
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/AbstractSqlRedisReceiver.java

+package com.liquidnet.service.consumer.base.receiver;
+
+import com.liquidnet.commons.lang.util.CollectionUtil;
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.base.SqlMapping;
+import com.liquidnet.service.consumer.base.service.IBaseDao;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.stream.MapRecord;
+import org.springframework.data.redis.connection.stream.StreamRecords;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.stream.StreamListener;
+
+import java.util.HashMap;
+
+/**
+ * 公共的SQL队列消息监听器，具体消费逻辑统一使用`consumerMessageHandler`
+ *
+ * @author zhanggb
+ * Created by IntelliJ IDEA at 2022/3/31
+ */
+@Slf4j
+public abstract class AbstractSqlRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
+    @Autowired
+    private IBaseDao baseDao;
+    @Autowired
+    StringRedisTemplate stringRedisTemplate;
+
+    @Override
+    public void onMessage(MapRecord<String, String, String> message) {
+        String redisStreamKey = this.getRedisStreamKey();
+        log.debug("CONSUMER MSG_SQL[streamKey:{},messageId:{},stream:{},body:{}]",redisStreamKey, message.getId(), message.getStream(), message.getValue());
+        boolean result = this.consumerMessageHandler(message.getValue().get("message"));
+        log.info("CONSUMER MSG_SQL RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());
+
+        try {
+            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
+        } catch (Exception e) {
+            log.error("#CONSUMER MSG_SQL EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
+        }
+        try {
+            stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
+        } catch (Exception e) {
+            log.error("#CONSUMER MSG_SQL EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
+        }
+    }
+
+    private boolean consumerMessageHandler(String msg) {
+        boolean aBoolean = false;
+        try {
+            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msg, SqlMapping.SqlMessage.class);
+            aBoolean = null == sqlMessage || baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+        } catch (Exception e) {
+            log.error("CONSUMER MSG_SQL EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), msg, e);
+        } finally {
+            if (!aBoolean) {
+                HashMap<String, String> map = CollectionUtil.mapStringString();
+                map.put("message", msg);
+                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
+            }
+        }
+        return aBoolean;
+    }
+
+    protected abstract String getRedisStreamKey();
+
+    protected abstract String getRedisStreamGroup();
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/receiver/ConsumerCommonSQL0Receiver.java

+package com.liquidnet.service.consumer.base.receiver;
+
+import com.liquidnet.service.base.constant.MQConst;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ConsumerCommonSQL0Receiver extends AbstractSqlRedisReceiver {
+
+    @Override
+    protected String getRedisStreamKey() {
+        return MQConst.CommonQueue.SQL0.getKey();
+    }
+
+    @Override
+    protected String getRedisStreamGroup() {
+        return MQConst.CommonQueue.SQL0.getGroup();
+    }
+}
\ No newline at end of file

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/service/IBaseDao.java

+package com.liquidnet.service.consumer.base.service;
+
+import java.util.LinkedList;
+
+public interface IBaseDao {
+    /**
+     * 批量执行sql
+     *
+     * @param sql
+     * @param values
+     * @return
+     */
+    Boolean batchSql(String sql, LinkedList<Object[]> values);
+
+    /**
+     * 批量执行不定量sql
+     *
+     * @param sql
+     * @param values
+     * @return
+     */
+    Boolean batchSqls(LinkedList<String> sql, LinkedList<Object[]>... values);
+
+    /**
+     * 执行sql语句 无 参数
+     *
+     * @param sql
+     * @return
+     */
+    Boolean batchSqlNoArgs(LinkedList<String> sql);
+
+    /**
+     * xs 新增一条记录且返回主键Id
+     *
+     * @param sql   新增待执行sql
+     * @param param 参数
+     * @return 主键ID
+     */
+    int insertSqlAndReturnKeyId(final String sql, final Object[] param);
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/service/impl/BaseDao.java

+package com.liquidnet.service.consumer.base.service.impl;
+
+import com.liquidnet.commons.lang.util.JsonUtils;
+import com.liquidnet.service.consumer.base.service.IBaseDao;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.PreparedStatementCreator;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.jdbc.support.GeneratedKeyHolder;
+import org.springframework.jdbc.support.KeyHolder;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.TransactionStatus;
+import org.springframework.transaction.support.TransactionCallback;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import javax.annotation.Resource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.LinkedList;
+
+@Service
+public class BaseDao implements IBaseDao {
+    private static final Logger log = LoggerFactory.getLogger(BaseDao.class);
+    @Resource
+    public JdbcTemplate jdbcTemplate;
+    @Resource(name = "transactionManager")
+    public DataSourceTransactionManager transactionManager;
+
+    @Override
+    public Boolean batchSql(final String sql, final LinkedList<Object[]> values) {
+        TransactionCallback<Boolean> callback = new TransactionCallback<Boolean>() {
+            @Override
+            public Boolean doInTransaction(final TransactionStatus transactionStatus) {
+                if (values.size() > 0) {
+                    int[] ints = jdbcTemplate.batchUpdate(sql, values);
+                }
+                return true;
+            }
+        };
+        try {
+            TransactionTemplate tt = new TransactionTemplate(transactionManager);
+            return tt.execute(callback);
+        } catch (Exception ex) {
+            log.error("###\nSQL.Preparing:{}\nParameters:{}", JsonUtils.toJson(sql), JsonUtils.toJson(values), ex);
+            return false;
+        }
+    }
+
+    @Override
+    public Boolean batchSqls(final LinkedList<String> sql,
+                             final LinkedList<Object[]>... values) {
+        try {
+            TransactionCallback<Boolean> callback = new TransactionCallback<Boolean>() {
+                @Override
+                public Boolean doInTransaction(final TransactionStatus transactionStatus) {
+                    int i = 0;
+                    for (LinkedList<Object[]> o : values) {
+                        if (sql.size() < i + 1) {
+                            break;
+                        }
+                        if (!o.isEmpty()) {
+                            jdbcTemplate.batchUpdate(sql.get(i), o);
+                        }
+                        i++;
+                    }
+                    return true;
+                }
+            };
+
+            TransactionTemplate tt = new TransactionTemplate(transactionManager);
+            return tt.execute(callback);
+        } catch (Exception ex) {
+//            if (ex instanceof LiquidnetServiceException) {
+//                log.error("###Error.Code:{} - {}", ((LiquidnetServiceException) ex).getCode(), ex.getMessage());
+//            } else {
+            log.error("###Error.Sqls:{}\nParameters:{},Ex:{}", JsonUtils.toJson(sql), JsonUtils.toJson(values), ex.getMessage());
+//            }
+            return false;
+        }
+    }
+
+    @Override
+    public Boolean batchSqlNoArgs(final LinkedList<String> sql) {
+        try {
+            TransactionCallback<Boolean> callback = new TransactionCallback<Boolean>() {
+                @Override
+                public Boolean doInTransaction(final TransactionStatus transactionStatus) {
+                    for (String o : sql) {
+                        jdbcTemplate.execute(o);
+                    }
+                    return true;
+                }
+            };
+
+            TransactionTemplate tt = new TransactionTemplate(transactionManager);
+            return tt.execute(callback);
+        } catch (Exception ex) {
+            log.error("###Error.Sqls:{}\nParameters:{},Ex:{}", sql);
+            return false;
+        }
+    }
+
+    /**
+     * xs 新增一条记录且返回主键Id
+     *
+     * @param sql   新增待执行sql
+     * @param param 参数
+     * @return 主键ID
+     */
+    public int insertSqlAndReturnKeyId(final String sql, final Object[] param) {
+        final String innersql = sql;
+        final Object[] innerO = param;
+        KeyHolder keyHolder = new GeneratedKeyHolder();
+        try {
+            jdbcTemplate.update(new PreparedStatementCreator() {
+                @Override
+                public PreparedStatement createPreparedStatement(final Connection con)
+                        throws SQLException {
+                    PreparedStatement ps = con.prepareStatement(innersql,
+                            Statement.RETURN_GENERATED_KEYS);
+                    for (int i = 0; i < innerO.length; i++) {
+                        ps.setObject(i + 1, innerO[i]);
+                    }
+
+                    return ps;
+                }
+            }, keyHolder);
+        } catch (Exception e) {
+            log.error("###\nSQL.Preparing:{}\nParameters:{}", sql, JsonUtils.toJson(param), e);
+        }
+        return keyHolder.getKey().intValue();
+
+    }
+}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/service/processor/ConsumerAdamSmsProcessor.java

+//package com.liquidnet.service.consumer.base.service.processor;
+//
+//import com.liquidnet.common.mq.constant.MQConst;
+//import com.liquidnet.common.sms.processor.SmsProcessor;
+//import com.liquidnet.commons.lang.util.JsonUtils;
+//import com.liquidnet.service.base.SmsMessage;
+//import com.rabbitmq.client.Channel;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.amqp.core.Message;
+//import org.springframework.amqp.core.MessageProperties;
+//import org.springframework.amqp.rabbit.annotation.Exchange;
+//import org.springframework.amqp.rabbit.annotation.Queue;
+//import org.springframework.amqp.rabbit.annotation.QueueBinding;
+//import org.springframework.amqp.rabbit.annotation.RabbitListener;
+//import org.springframework.stereotype.Component;
+//
+//import javax.annotation.Resource;
+//import java.io.IOException;
+//
+///**
+// * ConsumerAdamSmsProcessor.class
+// *
+// * @author zhanggb
+// * Created by IntelliJ IDEA at 2021/7/13
+// */
+//@Slf4j
+//@Component
+//public class ConsumerAdamSmsProcessor {
+//    @Resource
+//    SmsProcessor smsProcessor;
+//
+//    private void consumerSmsSendHandler(Message msg, Channel channel) {
+//        MessageProperties properties = msg.getMessageProperties();
+//        String consumerQueue = properties.getConsumerQueue();
+//        long deliveryTag = properties.getDeliveryTag();
+//        log.info("CONSUMER SMS ==> [consumerQueue:{},deliveryTag:{}]", consumerQueue, deliveryTag);
+//        String msgBody = new String(msg.getBody());
+//        log.debug("CONSUMER SMS ==> Preparing:{}", msgBody);
+//        try {
+//            SmsMessage smsMessage = JsonUtils.fromJson(msgBody, SmsMessage.class);
+//            boolean result = smsProcessor.send(smsMessage.getPhone(), smsMessage.getSignName(), smsMessage.getTemplateCode(), smsMessage.getTemplateParam().toString());
+//            log.debug("CONSUMER SMS result of execution:{}", result);
+//            if (result) {
+//                channel.basicAck(deliveryTag, false);
+//            } else {
+//                log.warn("###CONSUMER SMS[consumerQueue:{},deliveryTag={},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody);
+//                channel.basicAck(deliveryTag, false);
+//            }
+//        } catch (IOException e) {
+//            log.error("CONSUMER SMS[consumerQueue:{},deliveryTag:{},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody, e);
+//        }
+//    }
+//
+//    /* ================================================================== ｜ 短信验证码 */
+//
+////    @RabbitListener(
+////            bindings = @QueueBinding(
+////                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
+////                    key = MQConst.RK_SMS_CODE,
+////                    value = @Queue(MQConst.QUEUES_SMS_CODE)
+////            ),
+////            concurrency = "25"
+////    )
+////    public void consumerSqlForSmsCode(Message msg, Channel channel) {
+////        this.consumerSmsSendHandler(msg, channel);
+////    }
+//
+//    /* ================================================================== ｜ 短信通知 */
+//
+//    @RabbitListener(
+//            bindings = @QueueBinding(
+//                    exchange = @Exchange(MQConst.EX_LNS_SMS_SENDER),
+//                    key = MQConst.RK_SMS_NOTICE,
+//                    value = @Queue(MQConst.QUEUES_SMS_NOTICE)
+//            ),
+//            concurrency = "10"
+//    )
+//    public void consumerSqlForSmsNotice(Message msg, Channel channel) {
+//        this.consumerSmsSendHandler(msg, channel);
+//    }
+//
+//
+//    /* ================================================================== ｜  */
+//}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/java/com/liquidnet/service/consumer/base/service/processor/ConsumerAdamUCenterProcessor.java

+//package com.liquidnet.service.consumer.base.service.processor;
+//
+//import com.liquidnet.common.mq.constant.MQConst;
+//import com.liquidnet.commons.lang.util.JsonUtils;
+//import com.liquidnet.service.base.SqlMapping;
+//import com.liquidnet.service.consumer.adam.service.IBaseDao;
+//import com.rabbitmq.client.Channel;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.amqp.core.Message;
+//import org.springframework.amqp.core.MessageProperties;
+//import org.springframework.amqp.rabbit.annotation.Exchange;
+//import org.springframework.amqp.rabbit.annotation.Queue;
+//import org.springframework.amqp.rabbit.annotation.QueueBinding;
+//import org.springframework.amqp.rabbit.annotation.RabbitListener;
+//import org.springframework.stereotype.Component;
+//
+//import javax.annotation.Resource;
+//import java.io.IOException;
+//
+///**
+// * ConsumerAdamProcessor.class
+// *
+// * @author zhanggb
+// * Created by IntelliJ IDEA at 2021/4/29
+// */
+//@Slf4j
+//@Component
+//public class ConsumerAdamUCenterProcessor {
+//    @Resource
+//    IBaseDao baseDao;
+//
+////    @RabbitListener(bindings = @QueueBinding(
+////            exchange = @Exchange(MQConst.EXCHANGES_LIQUIDNET_SQL), key = MQConst.ROUTING_KEY_SQL,
+////            value = @Queue(MQConst.QUEUES_SQL_MAIN)
+////    ))
+//////    @RabbitListener(queues = MQConst.QUEUES_SQL_MAIN)
+////    public void consumerSql(Message msg, Channel channel) {
+////        SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(new String(msg.getBody()), SqlMapping.SqlMessage.class);
+////        log.debug("CONSUMER SQL ==> Preparing:{}", JsonUtils.toJson(sqlMessage.getSqls()));
+////        log.debug("CONSUMER SQL ==> Parameters:{}", JsonUtils.toJson(sqlMessage.getArgs()));
+////
+////        try {
+////            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+////            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
+////            if (rstBatchSqls) {
+////                channel.basicAck(msg.getMessageProperties().getDeliveryTag(), false);
+////            } else {
+////                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+////            }
+////        } catch (Exception e) {
+////            log.error("error:CONSUMER SQL:{}", JsonUtils.toJson(sqlMessage), e);
+////            try {
+////                channel.basicReject(msg.getMessageProperties().getDeliveryTag(), true);
+////            } catch (IOException ioException) {
+////                log.error("error:CONSUMER SQL:basicReject.msg.tag:{}", msg.getMessageProperties().getDeliveryTag(), ioException);
+////            }
+////        }
+////    }
+//
+//    private void consumerSqlDaoHandler(Message msg, Channel channel) {
+//        MessageProperties properties = msg.getMessageProperties();
+//        String consumerQueue = properties.getConsumerQueue();
+//        long deliveryTag = properties.getDeliveryTag();
+//        log.info("CONSUMER SQL ==> [consumerQueue:{},deliveryTag:{}]", consumerQueue, deliveryTag);
+//        String msgBody = new String(msg.getBody());
+//        log.debug("CONSUMER SQL ==> Preparing:{}", msgBody);
+//        try {
+//            SqlMapping.SqlMessage sqlMessage = JsonUtils.fromJson(msgBody, SqlMapping.SqlMessage.class);
+//            Boolean rstBatchSqls = baseDao.batchSqls(sqlMessage.getSqls(), sqlMessage.getArgs());
+//            log.debug("CONSUMER SQL result of execution:{}", rstBatchSqls);
+//            if (rstBatchSqls) {
+//                channel.basicAck(deliveryTag, false);
+//            } else {
+//                log.warn("###CONSUMER SQL[consumerQueue:{},deliveryTag={},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody);
+//                channel.basicAck(deliveryTag, false);
+//            }
+//        } catch (IOException e) {
+//            log.error("CONSUMER SQL[consumerQueue:{},deliveryTag:{},sqlMessage:{}]", consumerQueue, deliveryTag, msgBody, e);
+//        }
+//    }
+//
+//    /* ================================================================== ｜ 用户注册 */
+//
+//    @RabbitListener(
+//            bindings = @QueueBinding(
+//                    exchange = @Exchange(MQConst.EX_LNS_SQL_UCENTER),
+//                    key = MQConst.RK_SQL_UREGISTER,
+//                    value = @Queue(MQConst.QUEUES_SQL_UREGISTER)
+//            ),
+//            concurrency = "5"
+//    )
+//    public void consumerSqlForURegister(Message msg, Channel channel) {
+//        this.consumerSqlDaoHandler(msg, channel);
+//    }
+//
+//    /* ================================================================== ｜ 用户信息 */
+//
+//    @RabbitListener(
+//            bindings = @QueueBinding(
+//                    exchange = @Exchange(MQConst.EX_LNS_SQL_UCENTER),
+//                    key = MQConst.RK_SQL_UCENTER,
+//                    value = @Queue(MQConst.QUEUES_SQL_UCENTER)
+//            ),
+//            concurrency = "5"
+//    )
+//    public void consumerSqlForUCenter(Message msg, Channel channel) {
+//        this.consumerSqlDaoHandler(msg, channel);
+//    }
+//
+//    /* ================================================================== ｜ 会员购买 */
+//
+//    @RabbitListener(
+//            bindings = @QueueBinding(
+//                    exchange = @Exchange(MQConst.EX_LNS_SQL_UCENTER),
+//                    key = MQConst.RK_SQL_UMEMBER,
+//                    value = @Queue(MQConst.QUEUES_SQL_UMEMBER)
+//            ),
+//            concurrency = "5"
+//    )
+//    public void consumerSqlForUMember(Message msg, Channel channel) {
+//        this.consumerSqlDaoHandler(msg, channel);
+//    }
+//}

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/resources/bootstrap-dev.yml

+# begin-dev-这里是配置信息基本值
+liquidnet:
+  cloudConfig:
+    profile: dev
+  security:
+    username: user
+    password: user123
+  eureka:
+    host: 39.107.71.112:7001
+# end-dev-这里是配置信息基本值
+
+spring:
+  profiles:
+    include: service-consumer-base

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/resources/bootstrap-prod.yml

+# begin-prod-这里是配置信息基本值
+liquidnet:
+  cloudConfig:
+    profile: prod
+  security:
+    username: user
+    password: user123
+  eureka:
+    host: 172.17.207.189:7001
+# end-prod-这里是配置信息基本值
+
+spring:
+  profiles:
+    include: service-consumer-base
\ No newline at end of file

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/resources/bootstrap-service-consumer-base.yml

+#eurekaServer配置
+eureka:
+  client:
+    register-with-eureka: true
+    fetch-registry: true
+    serviceUrl:
+      defaultZone: http://${liquidnet.security.username}:${liquidnet.security.password}@${liquidnet.eureka.host}/eureka-server/eureka
+#configServer配置
+spring:
+  cloud:
+    config:
+#      uri: http://39.107.71.112:7002/support-config
+      profile: ${liquidnet.cloudConfig.profile}
+      name: ${spring.application.name} #默认为spring.application.name
+      discovery:
+        enabled: true
+        service-id: liquidnet-support-config

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/resources/bootstrap-test.yml

+# begin-test-这里是配置信息基本值
+liquidnet:
+  cloudConfig:
+    profile: test
+  security:
+    username: user
+    password: user123
+  eureka:
+    host: 172.17.207.177:7001
+    #instance:
+    #  prefer-ip-address: true
+    #host: eureka-test-0.eureka-test-svc.zhengzai-test:7001/eureka-server/eureka,eureka-test-1.eureka-test-svc.zhengzai-test:7001/eureka-server/eureka,eureka-test-2.eureka-test-svc.zhengzai-test:7001/eureka-server/eureka
+    #host: 192.168.193.41:7001
+# end-test-这里是配置信息基本值
+
+spring:
+  profiles:
+    include: service-consumer-base
\ No newline at end of file

liquidnet-bus-service/liquidnet-service-consumer-all/liquidnet-service-consumer-base/src/main/resources/bootstrap.yml

+spring:
+  application:
+    name: liquidnet-service-consumer-base
+  profiles:
+    active: dev
\ No newline at end of file

liquidnet-bus-service/liquidnet-service-consumer-all/pom.xml

@@ -21,6 +21,7 @@
         <module>liquidnet-service-consumer-goblin</module>
         <module>liquidnet-service-consumer-slowly</module>
         <module>liquidnet-service-consumer-nft</module>
+        <module>liquidnet-service-consumer-base</module>
     </modules>
 
     <dependencies>