java代码使用RocketMQ进行发送与接收消息

一时失言乱红尘 2024-03-31 12:52 29阅读 0赞

> 消息队列RocketMQ版提供三种方式来发送普通消息：同步（Sync）发送、异步（Async）发送和单向（Oneway）发送。本文介绍了每种发送方式的使用方式。

#### 1.在pom.xml引入依赖 ####

<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.9.0</version>
          </dependency>

#### 2.创建测试类，使用同步发送 ####

//同步发送，发送消息成功之后，在收到服务端的相应后，才会继续发送下一条消息
        //应用场景：发邮件
        public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
    
            //RocketMQ 分很多消息发送与接收
            // 1同步 2异步 3单向
            //无论哪一种发送消息的类型，基本上一样
            //1.创建一个提供消息的对象，准备发送消息
            //2.通过nameserver对象，找到我们要往哪里发送消息（IP 端口）
            //3.创建消息（Topic Message key Tag）
            //4.通过nameServer找到broker后发送消息
            //5.关闭各种对象
    
            DefaultMQProducer producer = new DefaultMQProducer("tm_XBL");
            // RocketMQ 的ip地址已经端口号
            producer.setNamesrvAddr("192.168.1.7:9876");
            //需要将producer 启动起来 (这里有异常可测试阶段直接向上抛啦)
            producer.start();
            //编辑消息
            String body="{userName:'Tom',hobby:'爱吃白菜'}";
            //创建消息
            Message message=new Message("topicTmXBL","tagsTmXBL","keysTmXBL",body.getBytes());
            //发送消息  (这里有异常可测试阶段直接向上抛啦)
            SendResult send = producer.send(message);
            //返回结果查看
            System.out.println(send);
            //关闭链接
            producer.shutdown();
        }

控制台输出结果：ok

![2a54d34093794fbfbf6b5d7b2b6f42f4.png][]

#### 3.使用异步发送 ####

//异步发送消息
        //一般用在发送方不用等待服务器第一时间的响应，可以直接发送下一条消息
        //应用场景：用户上传身份证进行验证信息
        //等待到什么时候服务器审核到了用户发送过来的身份信息
        //再进行回调
        public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException {
    
            //前两步不变
            DefaultMQProducer producer = new DefaultMQProducer("tm_XBL");
            producer.setNamesrvAddr("192.168.1.7:9876");
            //需要将producer 对象启动
            producer.start();
            //编辑消息
            String body="{userName:'Jerry',hobby:'Tom'}";
            //创建消息
            Message message = new Message("topicTmXBL","tagsTmXBL","keysTmXBL",body.getBytes());
            //发送异步消息 发送结果通过callback返回给客户端
            producer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    //成功回调 同时关闭链接
                    System.out.println("回调成功");
                    producer.shutdown();
                }
    
                @Override
                public void onException(Throwable throwable) {
                    //失败回调 同时关闭链接
                    System.out.println("回调失败");
                    producer.shutdown();
                }
            });
    
            //不能关闭producer对象
            //因为producer要在后续代码执行的同时，同时监控成功/失败回调
    
        }

控制台结果：ok

![706ec8b0dc2349ed91e9932f3550ca4c.png][]

#### 4.单向发送 ####

//单向发送
        //只管发，不管早晚，也不管成败
        //应用场景：用于耗时短、且有可靠性，回调要求不高，例如 日志记录
        public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
    
            //前两步不变
            DefaultMQProducer producer = new DefaultMQProducer("tm_XBL");
            producer.setNamesrvAddr("192.168.1.7:9876");
            //需要将producer对象启动起来
            producer.start();
            //编辑信息
            String body="userName:'Allen',hobby:'Jerry'";
            //创建消息
            Message message=new Message("topicTmXBL","tagsTmXBL","keysTmXBL",body.getBytes());
            //发送信息
            producer.send(message);
            //关闭链接
            producer.shutdown();
        }

控制台结果：ok

![c96cc2d6d2994a4eae2bee7d43c34be0.png][]

#### 5.接收消息/消费消息 ####

//消费消息
        //注意，同一个消息可以被不同组的消费者消费
        //不能被同一组消费者重复消费
        public static void main(String[] args) throws MQClientException {
    
            //创建消费者对象
            //Push 是在消费者消费消息后可以返回告诉MQ消费状态的对象
            //Pull 是只管消费的对象，后续可能没什么处理
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tm_consumer");
            //通过NameSrv设置IP+端口
            consumer.setNamesrvAddr("192.168.1.7:9876");
            //设置一些消息的参数
            //是否顺序消费(先进先出/先进后出)
            //CONSUME_FROM_FIRST_OFFSET 先进先出
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //设置需要消费哪类信息
            consumer.subscribe("topicTmXBL","tagsTmXBL");
            //进行消费信息
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    //list就是根据条件得到的数据
                    //consumeOrderlyContext就是设置各种属性的上下文对象
                    //是否设置消费的消息在 被消费后被标记为以消费，相当于是删除的意思
                    consumeOrderlyContext.setAutoCommit(true);
                    list.forEach(a->{
                        String msgId = a.getMsgId();
                        String topic = a.getTopic();
                        String tags = a.getTags();
                        byte[] body = a.getBody();
                        String str=new String(body);
                        JSONObject jsonObject = JSON.parseObject(str);
                        System.out.print("用户姓名----"+jsonObject.get("userName")+",爱好----"+jsonObject.get("hobby")+",");
                        System.out.println(msgId+","+topic+","+tags);
                    });
    
    
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            //直接消费
            consumer.start();
    
        //    //消费者需要关闭吗？ 不用
        //    //1、我们在项目中使用消息消费者，是需要持续不断的读取MQ中的消息，
        //    //所以不用关闭
        //    //2、在读取消息时，是自动开启另一线程，和当前代码不是同时执行
        //    //可能造成，上面的consumeMessage方法还没走完，却已经consumer.shutdown了
        //    //consumer.shutdown();
    
        }

输出结果：

![81b0507cb9454df7baa814e84a762d6f.png][]

#### 最后总结一下三种发送方式的优缺点![fdeba1f30a9649228ba3610c39064967.png][] ####

[2a54d34093794fbfbf6b5d7b2b6f42f4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/be64cc30c33d4d6faa1b5a1c2d7e58fd.png
[706ec8b0dc2349ed91e9932f3550ca4c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/20033cd0628f45eba7f0180910e73dbb.png
[c96cc2d6d2994a4eae2bee7d43c34be0.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/8bb2d56d90ed41448b971d24422d5114.png
[81b0507cb9454df7baa814e84a762d6f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/86ef3b8a5aef47008da6298c0dc8c1f4.png
[fdeba1f30a9649228ba3610c39064967.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/31/e67748853fa44dcaace8d23c21713e47.png