consumer change to tcp

etl-core/src/main/java/com/tzld/crawler/etl/mq/EtlMQConsumer.java

@@ -41,66 +41,66 @@ import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
-/**
+///**
- * @author ehlxr
+// * @author ehlxr
- * @since 2023-06-09 15:24.
+// * @since 2023-06-09 15:24.
- */
+// */
-@Component
+//@Component
-public class EtlMQConsumer {
+//public class EtlMQConsumer {
-    private static final Logger log = LoggerFactory.getLogger(EtlMQConsumer.class);
+//    private static final Logger log = LoggerFactory.getLogger(EtlMQConsumer.class);
-    private final EtlService etlService;
+//    private final EtlService etlService;
-    @Value("${rocketmq.accessKey:}")
+//    @Value("${rocketmq.accessKey:}")
-    private String accessKey;
+//    private String accessKey;
-    @Value("${rocketmq.secretKey:}")
+//    @Value("${rocketmq.secretKey:}")
-    private String secretKey;
+//    private String secretKey;
-    @Value("${rocketmq.httpEndpoint:}")
+//    @Value("${rocketmq.httpEndpoint:}")
-    private String httpEndpoint;
+//    private String httpEndpoint;
-    @Value("${rocketmq.instanceId:}")
+//    @Value("${rocketmq.instanceId:}")
-    private String instanceId;
+//    private String instanceId;
-    @Value("${rocketmq.crawler.etl.topic:}")
+//    @Value("${rocketmq.crawler.etl.topic:}")
-    private String topic;
+//    private String topic;
-    @Value("${rocketmq.crawler.etl.groupid:}")
+//    @Value("${rocketmq.crawler.etl.groupid:}")
-    private String groupId;
+//    private String groupId;
-    @Value("${consumer.thread.size:10}")
+//    @Value("${consumer.thread.size:10}")
-    private Integer threadSize;
+//    private Integer threadSize;
-
+//
-    private MQConsumer consumer;
+//    private MQConsumer consumer;
-
+//
-    public EtlMQConsumer(EtlService etlService) {
+//    public EtlMQConsumer(EtlService etlService) {
-        this.etlService = etlService;
+//        this.etlService = etlService;
-    }
+//    }
-
+//
-    @PostConstruct
+//    @PostConstruct
-    public void init() {
+//    public void init() {
-        MQClient mqClient = new MQClient(httpEndpoint, accessKey, secretKey);
+//        MQClient mqClient = new MQClient(httpEndpoint, accessKey, secretKey);
-        consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
+//        consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
-
+//
-        ExecutorService service = Executors.newFixedThreadPool(threadSize);
+//        ExecutorService service = Executors.newFixedThreadPool(threadSize);
-        for (int i = 0; i < threadSize; i++) {
+//        for (int i = 0; i < threadSize; i++) {
-            service.submit(this::consumeMsg);
+//            service.submit(this::consumeMsg);
-        }
+//        }
-    }
+//    }
-
+//
-    private void consumeMsg() {
+//    private void consumeMsg() {
-        do {
+//        do {
-            try {
+//            try {
-                List<Message> messages = consumer.consumeMessage(1, 10);
+//                List<Message> messages = consumer.consumeMessage(1, 10);
-                if (messages == null || messages.isEmpty()) {
+//                if (messages == null || messages.isEmpty()) {
-                    log.info("No new message, continue");
+//                    log.info("No new message, continue");
-                    continue;
+//                    continue;
-                }
+//                }
-
+//
-                messages.forEach(message -> {
+//                messages.forEach(message -> {
-                    log.info("Receive message: {} from topic: {}, group: {}", message, topic, groupId);
+//                    log.info("Receive message: {} from topic: {}, group: {}", message, topic, groupId);
-                    CrawlerEtlParam param = JSONObject.parseObject(message.getMessageBodyString(), CrawlerEtlParam.class);
+//                    CrawlerEtlParam param = JSONObject.parseObject(message.getMessageBodyString(), CrawlerEtlParam.class);
-                    param.setMessageId(message.getMessageId());
+//                    param.setMessageId(message.getMessageId());
-                    etlService.deal(param);
+//                    etlService.deal(param);
-                    log.info("Deal done of message: {} from topic: {}, group: {}", message, topic, groupId);
+//                    log.info("Deal done of message: {} from topic: {}, group: {}", message, topic, groupId);
-                    consumer.ackMessage(Collections.singletonList(message.getReceiptHandle()));
+//                    consumer.ackMessage(Collections.singletonList(message.getReceiptHandle()));
-                });
+//                });
-            } catch (Throwable e) {
+//            } catch (Throwable e) {
-                log.error("Consume message from topic: {}, group: {} error", topic, groupId, e);
+//                log.error("Consume message from topic: {}, group: {} error", topic, groupId, e);
-            }
+//            }
-        } while (true);
+//        } while (true);
-    }
+//    }
-}
+//}