zoukankan      html  css  js  c++  java
  • rocketmq消费队列代码

            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(Constant.operationLogGroup);
            try {
                consumer.setNamesrvAddr(Constant.rocketQueneAddr);
                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                consumer.setMessageModel(MessageModel.BROADCASTING);
                consumer.subscribe(Constant.operationLogTopic, Constant.operationLogTag);
            } catch (MQClientException e) {
                logger.error("consume operation log MQ error", e);
            }
    
            cometutil = Comet4jUtil.getInstance(CHANNEL);
    
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    
                    byte[] bytes = msgs.get(0).getBody();
                    try {
                        cometutil.sendMesToAllConnsWithString(CHANNEL, new String(bytes, "UTF-8"));
                    } catch (UnsupportedEncodingException e) {
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            try {
                consumer.start();
                logger.info("operationLogController's MQ consumer started.");
            } catch (MQClientException e) {
                logger.error("consume operation log MQ start error", e);
            }
  • 相关阅读:
    Python os.getcwd()方法
    Python os.walk()方法
    PyTorch 模型构造
    Python map()函数
    字符串转数字测试--知识备忘
    如何判断一个变量是数组Array类型--实例--加个人见解
    js面试题
    ios学习笔记
    读取图片文件--实例
    名言记录
  • 原文地址:https://www.cnblogs.com/llguanli/p/8520437.html
Copyright © 2011-2022 走看看