RocketMQ发送同步异步单向消息

Posted 乐观男孩

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了RocketMQ发送同步异步单向消息相关的知识,希望对你有一定的参考价值。

目录

说明

后续打算根据RocketMQ官网的例子,也对各种类型的消息进行简单的使用。在熟悉基本使用,了解各类消息的特点之后,再抽时间进行内部原理剖析。后续所有的文章,都是基于RocketMQ-4.9.1 release版本。

基础工具类

在接下来的例子中,都会用到以下工具类RocketMqUtil,其主要作用是封装了RocketMQ的NameServer地址、Topic、生产者或消费者。所以这里先贴上该工具类的代码:

package com.lzj.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

public class RocketMqUtil 

    public static final String NAME_SERVER_ADDR = "192.168.6.130:9876";
    public static final String TOPIC = "topic_simple_message";

    public static DefaultMQProducer getDefaultMQProducer() 
        try 
            DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
            defaultMQProducer.setProducerGroup("TEST_PRODUCER");
            defaultMQProducer.setNamesrvAddr(NAME_SERVER_ADDR);
            //同步发送消息时发送失败后的重试次数
            defaultMQProducer.setRetryTimesWhenSendFailed(0);
            //异步发送消息时发送失败后的重试次数
            defaultMQProducer.setRetryTimesWhenSendAsyncFailed(0);
            defaultMQProducer.start();
            return defaultMQProducer;
         catch (Exception e) 
            e.printStackTrace();
            throw new RuntimeException("初始化RocketMQ生产者时发生异常");
        
    

    public static DefaultMQPushConsumer getDefaultMQPushConsumer() throws MQClientException 
        try 
            DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer();
            defaultMQPushConsumer.setConsumerGroup("TEST_CONSUMER");
            defaultMQPushConsumer.setNamesrvAddr(NAME_SERVER_ADDR);
            return defaultMQPushConsumer;
         catch (Exception e) 
            e.printStackTrace();
            throw new RuntimeException("初始化RocketMQ消费者时发生异常");
        
    


测试代码

由于代码比较简单,所以直接了当,上源码:

package com.lzj.rocketmq;

import lombok.extern.slf4j.Slf4j;
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.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
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.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

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

@SpringBootTest
@Slf4j
public class SimpleMessageTest 

    /**
     * 发送同步消息
     * @throws InterruptedException
     * @throws RemotingException
     * @throws MQClientException
     * @throws MQBrokerException
     */
    @Test
    public void sendMessageSynchronously() throws InterruptedException, RemotingException, MQClientException, MQBrokerException 
        DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
        Message message = new Message(RocketMqUtil.TOPIC, "send_message_synchronously", "同步消息".getBytes(Charset.forName("UTF-8")));
        SendResult sendResult = defaultMQProducer.send(message);
        log.info("同步消息,结果名称:,结果值:", sendResult.getSendStatus().name(),
                sendResult.getSendStatus().ordinal());
        defaultMQProducer.shutdown();
    

    /**
     * 发送异步消息
     * @throws RemotingException
     * @throws MQClientException
     * @throws InterruptedException
     */
    @Test
    public void sendMessageAsynchronously() throws RemotingException, MQClientException, InterruptedException 
        DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
        Message message = new Message(RocketMqUtil.TOPIC, "send_message_asynchronously", "异步消息".getBytes(Charset.forName("UTF-8")));
        defaultMQProducer.send(message, new SendCallback()

            @Override
            public void onSuccess(SendResult sendResult) 
                log.info("send message success." + sendResult.getSendStatus().ordinal());
            

            @Override
            public void onException(Throwable throwable) 
                log.info("send message fail.");
            
        );
        Thread.sleep(5000L);
        defaultMQProducer.shutdown();
    

    /**
     * 单向传输,客户端只管发送,不等待响应,存在一定的不安全性,一般使用于记录日志的功能
     * @throws RemotingException
     * @throws MQClientException
     * @throws InterruptedException
     */
    @Test
    public void sendMessageOneway() throws RemotingException, MQClientException, InterruptedException 
        DefaultMQProducer defaultMQProducer = RocketMqUtil.getDefaultMQProducer();
        Message message = new Message(RocketMqUtil.TOPIC, "send_message_oneway", "单向消息".getBytes(Charset.forName("UTF-8")));
        defaultMQProducer.sendOneway(message);
        defaultMQProducer.shutdown();
    

    @Test
    public void consumeMessage() throws Exception 
        DefaultMQPushConsumer defaultMQPushConsumer = RocketMqUtil.getDefaultMQPushConsumer();
        defaultMQPushConsumer.subscribe(RocketMqUtil.TOPIC, "*");
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() 
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) 
                log.info("消费到消息条数:", list.size());
                list.stream().map(messageExt -> new String(messageExt.getBody(), Charset.forName("UTF-8")))
                        .map(String::new).forEach(System.out::println);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            
        );
        defaultMQPushConsumer.start();
        Thread.sleep(5000L);
    




这里展示了三类基础消息的发送及消费的例子:

**同步消息:**客户端将消息发送到Master Broker后,需要等Master Broker将消息同步到对应的Slave Broker后,Broker才会响应客户端。这种方式响应客户端比较慢,但具有较强的数据安全性,数据丢失风险小。

**异步消息:**客户端将消息发送到Master Broker后,只要Master Broker将消息处理完,Broker就响应客户端,无需等待Master Broker将消息同步到对应的Slave Broker。这种方式对客户端响应快,但存在一定的风险性:在Master Broker处理完消息响应客户端后就宕机,此时消息可能会发生丢失。

**单向消息:**客户端只负责消息发送,不等待Broker回应,这种方式效率最高,但消息丢失的风险也最高,一般用在可以容忍消息有一定丢失的场景。

效果展示

从RocketMQ控制台上看到消息确实发送成功(测试时同步消息发了两次)

源码消费者也同样的成功消费到数据:

以上是关于RocketMQ发送同步异步单向消息的主要内容,如果未能解决你的问题,请参考以下文章

RocketMQ源码—Producer发送消息源码—单向同步异步发送消息一万字

RocketMQ源码—Producer发送单向同步异步消息源码一万字

RocketMQ 保证消息不丢失

rocketmq简单消息发送

RocketMQ源码—Producer发送消息的总体流程一万字

RocketMQ源码—Producer发送消息源码—发送消息的总体流程一万字