自学内容网 自学内容网
自学内容网 > 正文

rocketmq一个简单的发送消息接收消息

🕗 发布于 2024-09-25 20:45 rocketmq  

需要的依赖

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

发送消息

package com.zxy.bank2.test;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;

public class PorducerTest {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 1、创建DefaultMQProducer
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("pg_1");
        // 2、设置namesrvAddr
        defaultMQProducer.setNamesrvAddr("localhost:9876");
        // 3、开启DefaultMQProducer
        defaultMQProducer.start();
        // 4、创建消息Message
        Message message = new Message("topic_1"
                , "tags_1" // 主要用于消息过滤
                , "keys_1" // 消息的唯一值
                , "消息内容hello".getBytes(StandardCharsets.UTF_8));
        // 5、发送消息
        SendResult result = defaultMQProducer.send(message);
        System.out.println(result);
        // 6、关闭DefaultMQProducer
        defaultMQProducer.shutdown();
    }
}

消费消息

package com.zxy.bank2.test;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;

public class ConsumerTest {
    public static void main(String[] args) throws MQClientException {
        // 1、创建DefaultMQProducer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg_1");
        // 2、设置namesrvAddr
        consumer.setNamesrvAddr("localhost:9876");
        // 最大拉取消息个数
        consumer.setConsumeMessageBatchMaxSize(2);
        // 3、订阅消息的主题
        consumer.subscribe("topic_1", "*");
        // 4、创建消息监听Listener
        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                // 5、获取消息
                for (MessageExt messageExt : list) {
                    try {
                        // 获取主题
                        String topic = messageExt.getTopic();
                        // 获取标签
                        String tags = messageExt.getTags();
                        // 获取消息内容
                        String result = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        System.out.println("topic = " + topic);
                        System.out.println("tags = " + tags);
                        System.out.println("result = " + result);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        // 重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }

                }
                // 6、返回状态
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
}

运行结果：

rocketmq保证消息有序的方法是，发送消息时都发送到同一个队列，消费时就从这个队列获取消息，以保证消息的有序性。

发送消息到同一个队列的代码：

package com.zxy.bank2.test.order;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;
import java.util.List;

public class PorducerTest {
    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 1、创建DefaultMQProducer
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("pg_2");
        // 2、设置namesrvAddr
        defaultMQProducer.setNamesrvAddr("localhost:9876");
        // 3、开启DefaultMQProducer
        defaultMQProducer.start();

        // 5、发送消息。5条消息将发送到同一个队列里
        for (int i = 0; i < 5; i++) {
            // 4、创建消息Message
            Message message = new Message("topic_2"
                    , "tags_2" // 主要用于消息过滤
                    , "keys_" + i // 消息的唯一值
                    , ("消息内容hello" + i).getBytes(StandardCharsets.UTF_8));
            SendResult result = defaultMQProducer.send(
                    // 消息内容
                    message,
                    // 选择指定的消息队列对象
                    new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                            // 获取队列的下标
                            Integer index = (Integer) o;
                            // 获取对应下标的队列
                            return list.get(index);
                        }
                    },
                    // 指定对应的队列下标
                    1);
            System.out.println(result);
        }
        // 6、关闭DefaultMQProducer
        defaultMQProducer.shutdown();
    }
}

运行结果：

有序消息消费的代码：

package com.zxy.bank2.test.order;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;

public class ConsumerTest {
    public static void main(String[] args) throws MQClientException {
        // 1、创建DefaultMQProducer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg_2");
        // 2、设置namesrvAddr
        consumer.setNamesrvAddr("localhost:9876");
        // 最大拉取消息个数
        consumer.setConsumeMessageBatchMaxSize(2);
        // 3、订阅消息的主题
        consumer.subscribe("topic_2", "*");
        // 4、创建消息监听Listener
        consumer.setMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                // 5、获取消息
                for (MessageExt messageExt : list) {
                    try {
                        // 获取主题
                        String topic = messageExt.getTopic();
                        // 获取标签
                        String tags = messageExt.getTags();
                        // 获取消息内容
                        String result = new String(messageExt.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        System.out.println("topic = " + topic + "tags = " + tags + "result = " + result);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        // 重试
                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }

                }
                // 6、返回状态
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
    }
}

执行结果：


原文地址：https://blog.csdn.net/2301_76947341/article/details/142525941

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

相关文章

自学内容网
Copyright © 2015-2024