消息队列RocketMQ入门实践（二）

前言

嗨，大家好，我是希留。

上一篇文章中我们介绍了RocketMQ的相关概念，以及安装部署。消息队列RocketMQ入门实践（一）
今天我们就来具体操作一下，在代码里面如何实现发送消息，消费消息。

项目源码：Gitee地址

1、快速入门 1.1 创建topic

public class TopicDemo {

    public static void main(String[] args) throws MQClientException{
        // 创建一个生产者组
        DefaultMQProducer producer = new DefaultMQProducer("xiliu_producer_group");
        // 生产者需用通过NameServer获取所有broker的路由信息，多个用分号隔开
        producer.setNamesrvAddr("42.194.222.32:9876");
        // 启动
        producer.start();

        // 创建topic  key：broker名称  newTopic：topic名称  queueNum：队列数（分区）
        producer.createTopic("broker_xiliu", "xiliu_topic", 8);

        System.out.println("创建topic成功");

        producer.shutdown();
    }
}

运行出错：

原因：开发机上访问不到broker的ip地址。
所以，需要指定broker的ip地址。

#修改broker配置文件
vim /conf/broker.conf
brokerIP1=42.194.222.32
namesrvAddr=42.194.222.32:9876

修改完成之后，先关闭启动了的broker，再重新启动。

#关闭命令
sh bin/mqshutdown broker

#重新启动broker，通过 -c 指定配置文件
nohup sh bin/mqbroker -c conf/broker.conf  &

运行代码，创建topic成功

1.2 发送消息（同步）

代码示例如下：

public class SyncProducerDemo {

    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("xiliu_producer_group");
        producer.setNamesrvAddr("42.194.222.32:9876");
        producer.start();
        String msgStr = "用户A发送同步消息给用户B";
        
        Message message = new Message("broker-a","xiliu_tag",msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
        //SendResult是发送结果的封装，包括消息状态，消息id，选择的队列等等，只要不抛异常，就代表发送成功
        SendResult sendResult = producer.send(message, 10000);
        System.out.println("消息状态：" + sendResult.getSendStatus());
        System.out.println("消息id：" + sendResult.getMsgId());
        System.out.println("消息queue：" + sendResult.getMessageQueue());
        System.out.println("消息offset：" + sendResult.getQueueOffset());
        producer.shutdown();
    }
}

发送成功。

1.2.1 Message数据结构

1.3 发送消息（异步）

代码示例如下：

public class AsyncProducerDemo {

    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("xiliu_producer_group");
        producer.setNamesrvAddr("42.194.222.32:9876");
        // 发送失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(0);

        producer.start();

        Message msg = new Message("xiliu_topic","xiliu_async_tag","Hello 我是异步消息".getBytes(RemotingHelper.DEFAULT_CHARSET));

        final CountDownLatch countDownLatch = new CountDownLatch(1);
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("消息状态：" + sendResult.getSendStatus());
                System.out.println("消息id：" + sendResult.getMsgId());
                System.out.println("消息queue：" + sendResult.getMessageQueue());
                System.out.println("消息offset：" + sendResult.getQueueOffset());
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败！" + throwable);
                throwable.printStackTrace();
            }
        },10000);
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.shutdown();
    }
}

运行成功。

1.4 消费消息

代码示例如下：

public class ConsumerDemo {

    public static void main(String[] args) throws Exception{
        // 推送型消费者
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("xiliu_consumer_group");
        pushConsumer.setNamesrvAddr("42.194.222.32:9876");
        // 订阅topic，接收此Topic下的所有消息
        pushConsumer.subscribe("xiliu_topic", "*");

        pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                
                System.out.println("收到消息->" + msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        pushConsumer.start();
    }
}

1.5 消息过滤器

RocketMQ支持根据用户自定义属性进行过滤，过滤表达式类似于SQL的where，如：a> 5 AND b =‘abc’

发送消息：

public class SyncProducerFilterDemo {

    public static void main(String[] args) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer("xiliu_producer_group");
        producer.setNamesrvAddr("42.194.222.32:9876");
        producer.start();

        String msgStr = "我是希留001";
        Message message = new Message("broker-a","xiliu_tag",msgStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
        message.putUserProperty("age", "21");
        message.putUserProperty("sex", "男");
        // 发送消息
        SendResult sendResult = producer.send(message,10000);

        System.out.println(sendResult);
        producer.shutdown();
    }
}

接收消息：

public class ConsumerFilterDemo {

    public static void main(String[] args) throws Exception{
        DefaultMQPushConsumer pushConsumer = new DefaultMQPushConsumer("xiliu_consumer_group");
        pushConsumer.setNamesrvAddr("42.194.222.32:9876");
        // 订阅topic，接收此Topic下的所有消息
        pushConsumer.subscribe("broker-a", MessageSelector.bySql("age>=20 AND sex='女'"));

        pushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("收到消息->" + msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        pushConsumer.start();
    }
}

运行出现报错：

原因是默认配置下，不支持自定义属性，需要设置开启：

#加入到broker的配置文件中
enablePropertyFilter=true

重启broker进行测试，发现已经可以接收消息了。

2、SpringBoot 整合 RocketMQ

apache 官方已经提供了 rocketmq 对应的 springboot starter，这极大的简化了我们所需要做的配置工作。

2.1 引入依赖

先在 pom.xml 中引入 apache 官方提供的 starter

    org.apache.rocketmq
    rocketmq-spring-boot-starter
    2.1.0

2.2 配置 application.yml

在 application.yml 配置一个 name-server 地址

rocketmq:
  name-server: 42.194.222.32:9876
  producer:
    group: xiliu_producer_group

2.3 创建一个生产者类

生产者的代码比较简单，只需要注入RocketMQTemplate就可以发送消息。

@Component
public class MessageSender {

    @Autowired
    private RocketMQTemplate rocketMqTemplate;


    public void syncSend() {
        
        SendResult result = rocketMqTemplate.syncSend("broker-a:xiliu_tag", "这是一条同步消息", 10000);
        System.out.println(result);
    }

    public void asyncSend() throws Exception {
        
        rocketMqTemplate.asyncSend("broker-a:xiliu_tag", "这是一条异步消息", new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功:" + sendResult);
            }
            @Override
            public void onException(Throwable e) {
                System.out.println("异步消息发送失败！" + e.getMessage());
            }
        });
        TimeUnit.SECONDS.sleep(100000);
    }

    public void sendOneWay() {
        
        rocketMqTemplate.sendOneWay("broker-a:xiliu_tag", "这是一条单向消息");
    }

    public void sendOneWayOrderly() {
        
        rocketMqTemplate.sendOneWayOrderly("broker-a:xiliu_tag", "这是一条顺序消息","888");

    }
}

2.4 创建一个消费者类

@Component
@RocketMQMessageListener(consumerGroup = "xiliu_consumer_group", topic = "broker-a",
        selectorexpression = "xiliu_tag", selectorType = SelectorType.TAG,
        messageModel = MessageModel.CLUSTERING, consumeMode = ConsumeMode.CONCURRENTLY)
public class MessageConsumer implements RocketMQListener{

    @Override
    public void onMessage(String s) {
        try {
            System.out.println("接收到rocketmq消息:" + s);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

MessageModel有两个选项，

BROADCASTING 代表所有消费者消费同样的消息CLUSTERING 代表多个消费者轮询消费消息（默认）。

ConsumeMode也有两个选项，

CONCURRENTLY代表消费端并发消费（默认），消息顺序得不到保证，到底有多少个线程并发消费，取决于线程池的大小。ORDERLY代表有序消费，也就是生产者发送的顺序跟消费者消费的顺序一致。 2.5 编写测试类

@SpringBootTest(classes = SpringbootRocketmqApplication.class)
public class RocketmqApplicationTest {

    @Autowired
    private MessageSender sender;

    @Test
    public void syncSendTest() {
        sender.syncSend();
    }


    @Test
    public void asyncSendTest() throws Exception {
        sender.asyncSend();
    }


    @Test
    public void sendOneWayTest() {
        sender.sendOneWay();
    }


    @Test
    public void sendOneWayOrderlyTest() {
        sender.sendOneWayOrderly();
    }

}

执行一下，查看结果，发送成功并且消费成功了。

总结

感谢大家的阅读，以上就是今天要讲的内容。本文简单介绍了RocketMQ发送消息、消费消息的简单示例，以及SpringBoot如何整合rocketmq。

感兴趣的小伙伴们，赶紧动动手来实践一下吧，若觉得对你有帮助的话，帮忙点赞评论关注，支持一波哟~