文档章节

RocketMQ入门案例

o
 osc_w9s1w4o0
发布于 2019/03/29 11:27
字数 1666
阅读 9
收藏 0

精选30+云产品,助力企业轻松上云!>>>

  学习RocketMQ,先写一个Demo演示一下看看效果。

一、服务端部署

  因为只是简单的为了演示效果,服务端仅部署单Master模式 —— 一个Name Server节点,一个Broker节点。主要有以下步骤。

    1. 下载RocketMQ源码、编译(也可以网上下载编译好的文件),这里使用最新的4.4.0版本,下载好之后放在Linux上通过一下命令解压缩、编译。
      unzip rocketmq-all-4.4.0-source-release.zip
      cd rocketmq-all-4.4.0/
      mvn -Prelease-all -DskipTests clean install –U

       

    2. 编译之后到distribution/target/apache-rocketmq目录,后续所有操作都是在该路径下。
      cd distribution/target/apache-rocketmq

       

    3. 启动Name Server,查看日志确认启动成功。
      nohup sh bin/mqnamesrv &
      tail -f ~/logs/rocketmqlogs/namesrv.log

       

    4. 启动Broker,查看日志确认启动成功。
      nohup sh bin/mqbroker -n localhost:9876 &
      tail -f ~/logs/rocketmqlogs/broker.log

       

  Name Server和Broker都成功启动,服务器就部署完成了。更详细的参考官方文档手册,里面还包含在服务器上运行Producer、Customer示例,这里主要是在项目中使用。

  官网手册戳这里:Quick Start

二、客户端搭建:Spring Boot项目中使用

  客户端分为消息生产者和消息消费者,这里通过日志打印输出查看效果,为了看起来更清晰,我新建了两个模块分别作为消息生产者和消息消费者。

  1. 添加依赖,在两个模块的pom文件中添加以下配置。
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>4.4.0</version>
    </dependency>
  2. 配置生产者模块。
    • application.yml文件中增加用来初始化producer的相关配置,这里只配了一部分,更详细的配置参数可以查看官方文档。
      # RocketMQ生产者
      rocketmq:
        producer:
          # Producer组名,多个Producer如果属于一个应用,发送同样的消息,则应该将它们归为同一组。默认DEFAULT_PRODUCER
          producerGroup: ${spring.application.name}
          # namesrv地址
          namesrvAddr: 192.168.101.213:9876
          # 客户端限制的消息大小,超过报错,同时服务端也会限制,需要跟服务端配合使用。默认4MB
          maxMessageSize: 4096
          # 发送消息超时时间,单位毫秒。默认10000
          sendMsgTimeout: 5000
          # 如果消息发送失败,最大重试次数,该参数只对同步发送模式起作用。默认2
          retryTimesWhenSendFailed: 2
          # 消息Body超过多大开始压缩(Consumer收到消息会自动解压缩),单位字节。默认4096
          compressMsgBodyOverHowmuch: 4096
          # 在发送消息时,自动创建服务器不存在的topic,需要指定Key,该Key可用于配置发送消息所在topic的默认路由。
          createTopicKey: XIAO_LIU

       

    • 新增producer配置类,系统启动时读取yml文件的配置信息初始化producer。集群模式下,如果在同一个jvm中,要往多个的MQ集群发送消息,则需要创建多个的producer并设置不同的instanceName,默认不需要设置该参数。
      @Configuration
      public class ProducerConfiguration {
          private static final Logger LOGGER = LoggerFactory.getLogger(ProducerConfiguration.class);
      
          /**
           * Producer组名,多个Producer如果属于一个应用,发送同样的消息,则应该将它们归为同一组。默认DEFAULT_PRODUCER
           */
          @Value("${rocketmq.producer.producerGroup}")
          private String producerGroup;
          /**
           * namesrv地址
           */
          @Value("${rocketmq.producer.namesrvAddr}")
          private String namesrvAddr;
          /**
           * 客户端限制的消息大小,超过报错,同时服务端也会限制,需要跟服务端配合使用。默认4MB
           */
          @Value("${rocketmq.producer.maxMessageSize}")
          private Integer maxMessageSize;
          /**
           * 发送消息超时时间,单位毫秒。默认10000
           */
          @Value("${rocketmq.producer.sendMsgTimeout}")
          private Integer sendMsgTimeout;
          /**
           * 如果消息发送失败,最大重试次数,该参数只对同步发送模式起作用。默认2
           */
          @Value("${rocketmq.producer.retryTimesWhenSendFailed}")
          private Integer retryTimesWhenSendFailed;
          /**
           * 消息Body超过多大开始压缩(Consumer收到消息会自动解压缩),单位字节。默认4096
           */
          @Value("${rocketmq.producer.compressMsgBodyOverHowmuch}")
          private Integer compressMsgBodyOverHowmuch;
          /**
           * 在发送消息时,自动创建服务器不存在的topic,需要指定Key,该Key可用于配置发送消息所在topic的默认路由。
           */
          @Value("${rocketmq.producer.createTopicKey}")
          private String createTopicKey;
      
          @Bean
          public DefaultMQProducer getRocketMQProducer() {
      
              DefaultMQProducer producer = new DefaultMQProducer(this.producerGroup);
              producer.setNamesrvAddr(this.namesrvAddr);
              producer.setCreateTopicKey(this.createTopicKey);
      
              if (this.maxMessageSize != null) {
                  producer.setMaxMessageSize(this.maxMessageSize);
              }
              if (this.sendMsgTimeout != null) {
                  producer.setSendMsgTimeout(this.sendMsgTimeout);
              }
              if (this.retryTimesWhenSendFailed != null) {
                  producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
              }
              if (this.compressMsgBodyOverHowmuch != null) {
                  producer.setCompressMsgBodyOverHowmuch(this.compressMsgBodyOverHowmuch);
              }
              if (Strings.isNotBlank(this.createTopicKey)) {
                  producer.setCreateTopicKey(this.createTopicKey);
              }
      
              try {
                  producer.start();
      
                  LOGGER.info("Producer Started : producerGroup:[{}], namesrvAddr:[{}]"
                          , this.producerGroup, this.namesrvAddr);
              } catch (MQClientException e) {
                  LOGGER.error("Producer Start Failed : {}", e.getMessage(), e);
              }
              return producer;
          }
      
      }

       

    • 使用producer实例向MQ发送消息。
      @RunWith(SpringRunner.class)
      @SpringBootTest
      public class ProducerServiceApplicationTests {
          private static final Logger LOGGER = LoggerFactory.getLogger(ProducerServiceApplicationTests.class);
          @Autowired
          private DefaultMQProducer defaultMQProducer;
      
          @Test
          public void send() throws MQClientException, RemotingException, MQBrokerException, InterruptedException, UnsupportedEncodingException {
              for (int i = 0; i < 100; i++) {
                  User user = new User();
                  user.setUsername("用户" + i);
                  user.setPassword("密码" + i);
                  user.setSex(i % 2);
                  user.setBirthday(new Date());
                  Message message = new Message("user-topic", "user-tag", JSON.toJSONString(user).getBytes(RemotingHelper.DEFAULT_CHARSET));
                  SendResult sendResult = defaultMQProducer.send(message);
                  LOGGER.info(sendResult.toString());
              }
          }
      }

       

  3. 配置消费者模块。
    • application.yml文件中增加用来初始化consumer的相关配置,同样参数这里只配了一部分,更详细的配置参数可以查看官方文档。
      # RocketMQ消费者
      rocketmq:
        consumer:
          # Consumer组名,多个Consumer如果属于一个应用,订阅同样的消息,且消费逻辑一致,则应该将它们归为同一组。默认DEFAULT_CONSUMER
          consumerGroup: ${spring.application.name}
          # namesrv地址
          namesrvAddr: 192.168.101.213:9876
          # 消费线程池最大线程数。默认10
          consumeThreadMin: 10
          # 消费线程池最大线程数。默认20
          consumeThreadMax: 20
          # 批量消费,一次消费多少条消息。默认1
          consumeMessageBatchMaxSize: 1
          # 批量拉消息,一次最多拉多少条。默认32
          pullBatchSize: 32
          # 订阅的主题
          topics: user-topic

       

    • 新增consumer配置。
      @Configuration
      public class ConsumerConfiguration {
          private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerConfiguration.class);
      
          @Value("${rocketmq.consumer.consumerGroup}")
          private String consumerGroup;
          @Value("${rocketmq.consumer.namesrvAddr}")
          private String namesrvAddr;
          @Value("${rocketmq.consumer.consumeThreadMin}")
          private int consumeThreadMin;
          @Value("${rocketmq.consumer.consumeThreadMax}")
          private int consumeThreadMax;
          @Value("${rocketmq.consumer.consumeMessageBatchMaxSize}")
          private int consumeMessageBatchMaxSize;
          @Value("${rocketmq.consumer.pullBatchSize}")
          private int pullBatchSize;
          @Value("${rocketmq.consumer.topics}")
          private String topics;
      
          private final ConsumeMsgListener consumeMsgListener;
      
          @Autowired
          public ConsumerConfiguration(final ConsumeMsgListener consumeMsgListener) {
              this.consumeMsgListener = consumeMsgListener;
          }
      
          @Bean
          public DefaultMQPushConsumer getRocketMQConsumer() {
              DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
              consumer.setNamesrvAddr(namesrvAddr);
              consumer.setConsumeThreadMin(consumeThreadMin);
              consumer.setConsumeThreadMax(consumeThreadMax);
              consumer.setConsumeMessageBatchMaxSize(consumeMessageBatchMaxSize);
              consumer.setPullBatchSize(pullBatchSize);
              consumer.registerMessageListener(consumeMsgListener);
      
              try {
                  /**
                   * 设置消费者订阅的主题和tag。subExpression参数为*表示订阅该主题下所有tag,
                   * 如果需要订阅该主题下的指定tag,subExpression设置为对应tag名称,多个tag以||分割,例如"tag1 || tag2 || tag3"
                   */
                  consumer.subscribe(topics, "*");
                  consumer.start();
      
                  LOGGER.info("Consumer Started : consumerGroup:{}, topics:{}, namesrvAddr:{}", consumerGroup, topics, namesrvAddr);
              } catch (Exception e) {
                  LOGGER.error("Consumer Start Failed : consumerGroup:{}, topics:{}, namesrvAddr:{}", consumerGroup, topics, namesrvAddr, e);
                  e.printStackTrace();
              }
              return consumer;
          }
      }

       

    • 新增消息监听器,监听到新消息后,执行对应的业务逻辑。

      @Component
      public class ConsumeMsgListener implements MessageListenerConcurrently {
          private static final Logger LOGGER = LoggerFactory.getLogger(ConsumeMsgListener.class);
      
          @Override
          public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
              if (CollectionUtils.isEmpty(msgs)) {
                  LOGGER.info("Msgs is Empty.");
                  return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
              }
              for (MessageExt msg : msgs) {
                  try {
                      if ("user-topic".equals(msg.getTopic())) {
                          LOGGER.info("{} Receive New Messages: {}", Thread.currentThread().getName(), new String(msg.getBody()));
                          // do something
                      }
                  } catch (Exception e) {
                      if (msg.getReconsumeTimes() == 3) {
                          // 超过3次不再重试
                          LOGGER.error("Msg Consume Failed.");
                          return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                      } else {
                          // 重试
                          return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                      }
                  }
              }
      
              return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
          }
      }

       

三、效果
  1. 运行生产者测试代码。系统启动时初始化Producer,然后执行测试代码,往MQ中发送消息。效果如下:

     

  2. 启动消费者服务。系统启动时先初始化Customer。此时1.已经往MQ中发送了一些消息,监听器监听到MQ中有消息,随即马上消费消息。

     

四、总结

  Demo很简单,但是里面还有很多东西需要慢慢研究。

  代码可以戳这里:spring-cloud-learn

上一篇: C#闭包
o
粉丝 0
博文 500
码字总数 0
作品 0
私信 提问
加载中
请先登录后再评论。
RocketMQ 的使用

项目github : https://github.com/alibaba/RocketMQ/ 下载:https://github.com/alibaba/RocketMQ/releases marco_tan系列文章: http://www.cnblogs.com/marcotan/category/655319.html 1......

小白001
2015/11/04
4
0
发力IOT、大数据,Apache RocketMQ 构建云时代的新生态

摘要 2018年9月1日,由阿里巴巴中间件举办的RocketMQ开发者沙龙在北京举行。这是RocketMQ今年举办的第二次大型线下技术交流活动,出席技术沙龙的嘉宾包括Apache RocketMQ创始人、OpenMessagi...

中间件小哥
2018/09/21
0
0
“Apache RocketMQ社区最佳贡献者评选”参选征集中

2019年,Apache RocketMQ开源生态得到了进一步壮大和发展。内核仓库先后发布了三个大版本,新增了消息轨迹、多副本存储、ACL、RPC模式、IPV6等重要特性,孵化了Connect、RocketMQ Operator、...

oschina
2019/12/25
2.1K
1
互联网架构师学习笔记整理

互联网架构师学习笔记整理-完善中 一、并发编程 + ActiveMQ + 实战案例 并发编程基础篇 第一天 1、课程大纲简要介绍 2、线程基础概念、线程安全概念、多个线程多个锁概念 3、对象锁的同步和异...

osc_kf98pg0d
01/21
1
0
2018年技术直播PPT干货分享

随着互联网技术的飞速发展,知识更新换代的频率也随着加快。为了实现知识共享,将最新的知识应用到实践领域。阿里专家汇集这里,给大家带来各类技术直播分享,涵盖人工智能、大数据、Java、数...

别捏燕燕脸
2018/09/12
0
0

没有更多内容

加载失败,请刷新页面

加载更多

将分段视频合并

环境 操作系统:Ubuntu Kylin 优麒麟 20.04 LTS 适用架构:AMD64、ARM64(鲲鹏、飞腾) 方法 将下载的视频分片段放入同一个文件夹。按片段排序的文件名汇入list.txt。 ls qq_video*.mp4 | s...

chipo
8分钟前
0
0
C#和Cpp混合编程系列1-内存管理

C#自带GC算法在独立使用时

梦想游戏人
12分钟前
6
0
SQL使用SELECT * [columnA除外] FROM tableA排除列?

问题: We all know that to select all columns from a table, we can use 我们都知道要从表中选择所有列,我们可以使用 SELECT * FROM tableA Is there a way to exclude column(s) from......

fyin1314
27分钟前
12
0
LVS-NAT模式示例配置

实验环境 Centos 7.6.1810 + ipvsadm v1.27 配置脚本 Director配置(1台) yum install -y ipvsadm; echo 1 > /proc/sys/net/ipv4/ip_forward; ipvsadm -C; # 清空所有集群配置 ipvsadm -A -......

月下狼
27分钟前
9
0
linux 复制 黏贴

飞雪无痕
37分钟前
19
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部