RokcetMQ之生产者消费者消息处理实例代码

╰半夏微凉° 2024-04-17 05:43 111阅读 0赞

首先创建springBoot项目，并引入依赖包

<dependency>
    		   <groupId>org.apache.rocketmq</groupId>
    		   <artifactId>rocketmq-client</artifactId>
    		   <version>4.4.0</version>
    		</dependency>
    		<dependency>
    		    <groupId>com.alibaba</groupId>
    		    <artifactId>fastjson</artifactId>
    		    <version>1.2.59</version>
    		</dependency>
    		<dependency>
    		    <groupId>org.slf4j</groupId>
    		    <artifactId>slf4j-api</artifactId>
    		    </dependency>
    		<dependency>
    		    <groupId>org.slf4j</groupId>
    		    <artifactId>slf4j-nop</artifactId>
    		</dependency>

# 生产者 #

## 发送消息的代码（同步消息及单向） ##

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.client.producer.SendStatus;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    public class ProducerTest {
    
    	public static void main(String[] args) throws Exception {
        	// 实例化消息生产者Producer 并确定生产者组
            DefaultMQProducer producer = new DefaultMQProducer("sayGroup");
        	// 设置NameServer的地址 多个那么server地址可以用分号; 隔开
        	producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
        	for (int i = 0; i < 10; i++) {
        	    // 创建消息，并指定Topic，Tag和消息体
        	    Message msg = new Message();
        	    msg.setTopic("TopicTest"); /* Topic */
        	    msg.setTags("TagA");/* Tag */
        	    msg.setBody(("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
        	    msg.setKeys("123456"); // key 业务层面的唯一标识码
                // msg.setDelayTimeLevel(3); // 设置延时等级3,这个消息将在10s之后发送
            	
                SendResult sendResult = producer.send(msg);// 发送消息到一个Broker （同步消息）
                // producer.sendOneway(msg); // 单向发送消息 没有响应，只有类似于日志系统不关心相应的才可能采用
                // 通过sendResult返回消息是否成功送达   SEND_OK 发送成功；
                System.out.println(sendResult);
                SendStatus sendStatus = sendResult.getSendStatus();
                // 判断结果
                if (sendStatus == SendStatus.SEND_OK) {
                    System.out.println("信息发送成功！");
                } else {
                    System.out.println("信息发送失败！");
                }
        	}
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }

## 异步发送消息 ##

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    
    public class AsyncProducer {
    
    	public static void main(String[] args) throws Exception {
        	// 实例化消息生产者Producer
            DefaultMQProducer producer = new DefaultMQProducer("say1Group");
        	// 设置NameServer的地址
            producer.setNamesrvAddr("localhost:9876");
        	// 启动Producer实例
            producer.start();
            producer.setRetryTimesWhenSendAsyncFailed(0);
        	for (int i = 0; i < 100; i++) {
                    final int index = i;
                	// 创建消息，并指定Topic，Tag和消息体
                    Message msg = new Message();
            	    msg.setTopic("TopicTest"); /* Topic */
            	    msg.setTags("TagA");/* Tag */
            	    msg.setBody(("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));/* Message body */
            	    msg.setKeys("123456"); // key 业务层面的唯一标识码
                    // SendCallback接收异步返回结果的回调
                    producer.send(msg, new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            System.out.printf("%-10d OK %s %n", index, sendResult.getMsgId());
                        }
                        @Override
                        public void onException(Throwable e) {
          	              System.out.printf("%-10d Exception %s %n", index, e);
          	              e.printStackTrace();
                        }
                	});
        	}
        	// 如果不再发送消息，关闭Producer实例。
        	producer.shutdown();
        }
    }

# 消费者消费 #

## push方式 ##

由MQ主动地将消息推送给消费者；采用Push方式，可以尽可能实时地将消息发送给消费者进行消费。但是，在消费者的处理消息的能力较弱的时候(比如，消费者端的业务系统处理一条消息的流程比较复杂，其中的调用链路比较多导致消费时间比较久。概括起来地说就是“慢消费问题”)，而MQ不断地向消费者Push消息，消费者端的缓冲区可能会溢出，导致异常；

import java.util.List;
    
    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;
    
    public class ConsumerPush {
    
    	public static void main(String[] args) throws InterruptedException, MQClientException {
    
        	// 实例化消费者
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("listenGroup");
    
        	// 设置NameServer的地址
            consumer.setNamesrvAddr("localhost:9876");
    
        	// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息  过滤tag可以使用 "TagA || TagC || TagD"  方式
            consumer.subscribe("TopicTest", "*");
        	// 注册监听类  回调实现类 MessageListenerConcurrently 来处理从broker拉取回来的消息 (Push模式)
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    // 标记该消息已经被成功消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动消费者实例
            consumer.start();
            System.out.println("Consumer Started");
    	}
    }

## pull ##

由消费者主动向MQ拉取消息；采用Pull方式，如何设置Pull消息的频率需要重点去考虑，举个例子来说，可能1分钟内连续来了1000条消息，然后2小时内没有新消息产生（概括起来说就是“消息延迟与忙等待”）。如果每次Pull的时间间隔比较久，会增加消息的延迟，即消息到达消费者的时间加长，MQ中消息的堆积量变大；若每次Pull的时间间隔较短，但是在一段时间内MQ中并没有任何消息可以消费，那么会产生很多无效的Pull请求的RPC开销，影响MQ整体的网络性能；

import java.util.HashMap;
    import java.util.Map;
    import java.util.Set;
    
    import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
    import org.apache.rocketmq.client.consumer.PullResult;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.message.MessageQueue;
    
    public class ConsumerPull {
    
    	//Java缓存
        private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();
        
    	public static void main(String[] args) throws InterruptedException, MQClientException {
    
    		// 消费者实例
    		DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("listenGroup");
            consumer.setNamesrvAddr("127.0.0.1:9876");
            consumer.setInstanceName("consumer"); // 客户端实例名称
            consumer.start();
    
            // 拉取订阅主题的队列，默认队列大小是4
            Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest");
            System.out.println("setSize: " + mqs.size());
            for (MessageQueue mq : mqs) {
                System.out.printf("Consume from the queue: %s%n", mq);
                SINGLE_MQ: // GOTO节点
                	while (true) {
                    try {
                         // 四个参数含义： mq, 从哪个队列中拉取消息;subExpression, SubscriptionData中的subString;offset, 消息拉取的offset;maxNums, 最大拉取的消息数目.
                        PullResult pullResult = consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32); // Consumer获取消息
                        System.out.printf("pullResult: %s%n", pullResult);
                        putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
                        switch (pullResult.getPullStatus()) {
                            case FOUND: // 发现消息
                                System.out.println(pullResult.getMsgFoundList().get(0).toString());
                                break;
                            case NO_NEW_MSG: // 没有新的消息
                                break SINGLE_MQ; // GOTO 
                            case NO_MATCHED_MSG: // 没有匹配的消息？？？
                            	
                            case OFFSET_ILLEGAL:
                                break;
                            default:
                                break;
                        }
                    } catch (Exception e) {
                    }
                }
            }
            consumer.shutdown();
    	}
    
    	private static void putMessageQueueOffset(MessageQueue mq, long offset) {
            offseTable.put(mq, offset);
        }
     
    	/**
    	 * 获取偏移量  消费者进度控制
    	 * @param mq
    	 * @return
    	 */
        private static long getMessageQueueOffset(MessageQueue mq) {
            Long offset = offseTable.get(mq);
            if (offset != null){
                System.out.println("offset:" + offset);
                return offset;
            }
            return 0;
        }
    }