栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 前沿技术 > 大数据 > 大数据系统

rocketmq SpringBoot生产者消费者代码及控制台入门

rocketmq SpringBoot生产者消费者代码及控制台入门

1.生产者
public class LogProducer {
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public String sendMessage(String msg) {
// 发送消息(参数1:主题,参数2:消息实体)
        SendResult res = rocketMQTemplate.syncSend(MqConfig.GP6_SYS_LOG_TOPIC, msg);
// 返回结果
        return SendStatus.SEND_OK.equals(res.getSendStatus()) ? PublicConst.SUCCESS : PublicConst.ERROR;
    }
}
// 定义的topic和消费者
public interface MqConfig {
    String GP6_SYS_LOG_TOPIC = "GP6_SYS_LOG_TOPIC";
    String GP6_SYS_LOG_CUSTOMER = "GP6_SYS_LOG_CUSTOMER";
}
1.1运行前:

没有该主题,也没有该消息。

1.2执行后:

新增的topic:

生产者:没有,因为我没定义

  消息:新增一条消息,点开后是hello world

2.消费者
@Slf4j
@Component
// 参数1:topic  参数2:消费者组
@RocketMQMessageListener(topic = MqConfig.GP6_SYS_LOG_TOPIC, consumerGroup = MqConfig.GP6_SYS_LOG_CUSTOMER)
public class LogConsumer implements RocketMQListener {
    @Resource
    private LogMongoService logService;

    @Override
    public void onMessage(String message) {
// 走到这就表示消息被消费了,可以直接去rocketMQ上看了。
        hanldMessage(message);
    }

// 下面可以写接口业务,如:存储消息到mongo中
    private void hanldMessage(String message) {
        if (null != message) {
            logService.saveOperationInfo(message);
            log.info("获取到的消息: " + message);
        }
    }
}

话外题:我们传入的数据:

 到这里拿到的是我们测试用例传入的参数。但是从界面看消息已经被消费了。

2.1运行后:

如下表示被消费:

新增消费者:

转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/710142.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号