RocketMQ 同步发送、异步发送和单向发送,如何选择?

RocketMQ 同步发送、异步发送和单向发送,如何选择?

你好,我是猿java。

在 RocketMQ 中,有 3种简单的消息发送方式:同步发送、异步发送和单向发送。这篇文章,我们将详细分析这三种发送方式的原理、优缺点、使用场景以及使用该方式是否会丢失数据。

本文源码基于: Apache RocketMQ release-5.2.0

同步发送

原理分析

在同步发送模式下,RocketMQ 默认采用同步刷盘方式,当生产者将消息发送到 Broker 后,会等待 Broker 的响应(默认超时 5分钟),Broker 接收消息后,会将其写入内存缓存,并进行刷盘操作。因此,如果 Broker 响应成功,代表消息一定成功写入磁盘。

g

同步发送主要涉及以下几个步骤:

创建Producer:创建一个Producer对象;

创建消息:创建一个Message对象,设置Topic、Tag标签和消息体;

发送消息:调用DefaultMQProducer的send方法;

等待响应:发送方会阻塞等待服务器的响应,直到收到确认消息;

如下示例代码为一个完整的同步发送流程:

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.client.producer.SendStatus;

import org.apache.rocketmq.common.message.Message;

public class SyncProducerTest {

public static void main(String[] args) throws Exception {

// 1、创建 producer,设置组名为 SyncGroupTest

DefaultMQProducer producer = new DefaultMQProducer("SyncGroup");

// 2、指定 NameServer的地址,以获取 Broker路由地址

producer.setNamesrvAddr("x.x.x.x:9876");

// 3、启动 producer

producer.start();

// 4、创建消息,并指定 Topic,Tag和消息体

Message msg = new Message("SyncTopic", "sync", "SyncMessage".getBytes("UTF-8"));

// 5、发送同步消息

SendResult sendResult = producer.send(msg);

// 6、通过 sendResult 判断消息是否成功送达

System.out.printf("message send result:" + sendResult);

// 7、关闭 Producer

producer.shutdown();

}

}

RocketMQ 的同步发送主要涉及以下几个关键源码类和方法:

DefaultMQProducer:生产者类,负责发送消息。

MQClientAPIImpl#sendMessage:底层消息发送实现。

NettyRemotingClient#invokeSync:通过 Netty 实现网络通信。

Broker 端的 SendMessageProcessor:处理发送请求。

源码参考:org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#send(Message msg)

优缺点

优点:

简单易用。

可靠性高,发送方可以确认消息是否成功发送,一旦发送成功,消息就已经写入磁盘,消息不会丢失。

缺点:

延迟较高,需要等待服务器的响应。

吞吐量可能受限于网络延迟和服务器性能。

使用场景

适用于对消息可靠性要求较高的场景,如订单系统、金融交易、重要的消息通知等。

异步发送

原理分析

在异步发送模式下,RocketMQ 默认采用异步刷盘方式,当生产者发送消息到 Broker 后,消息写入内存缓存成功后,Broker 立即返回响应(默认超时 5分钟),后台线程再异步将消息批量写入磁盘。因此,这种方式提高了系统的吞吐量和性能,但在系统崩溃时可能会丢失部分未刷盘的消息。

异步发送主要涉及以下几个步骤:

创建Producer:创建一个Producer对象;

创建消息:同样创建一个Message对象。

发送消息:调用DefaultMQProducer的send方法,但传递一个SendCallback回调对象。

处理响应:回调函数会在消息发送成功或失败时被调用。

如下示例代码为一个完整的异步发送流程:

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.client.producer.SendStatus;

import org.apache.rocketmq.common.message.Message;

public class AsyncProducerTest {

public static void main(String[] args) throws Exception {

// 1、创建 producer,设置组名为 AsyncGroupTest

DefaultMQProducer producer = new DefaultMQProducer("AsyncGroup");

// 2、指定 NameServer的地址,以获取 Broker路由地址

producer.setNamesrvAddr("x.x.x.x:9876");

// 3、启动 producer

producer.start();

// 4、创建消息,并指定Topic,Tag和消息体

Message msg = new Message("AsyncTopic","async", "AsyncMessage".getBytes("UTF-8"));

// 5、发送异步消息,SendCallback是处理异步回调的方法

producer.send(msg, new SendCallback() {

@Override

public void onSuccess(SendResult sendResult) {

// 成功回调

System.out.println("message send success: " + sendResult);

}

@Override

public void onException(Throwable throwable) {

// 失败回调

System.out.println("message send fail: " + throwable);

}

});

// 6、关闭 Producer

producer.shutdown();

}

}

RocketMQ 的异步发送主要涉及以下几个关键源码类和方法:

DefaultMQProducer:生产者类,负责发送消息。

MQClientAPIImpl#sendMessage:底层消息发送实现。

NettyRemotingClient#invokeAsync:通过 Netty 实现网络通信。

Broker 端的 SendMessageProcessor:处理发送请求。

源码参考:org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#send(Message msg, SendCallback sendCallback)

优缺点

优点:

非阻塞,发送方可以继续执行其他任务,提高吞吐量。

延迟较低,适用于对响应时间敏感的场景。

缺点:

实现复杂度较高,需要处理异步回调。

可靠性相对降低,需要处理失败重试等问题。

无法保证发送出去的数据不丢失。

使用场景

适用于对响应时间要求较高的场景,如实时数据处理、日志采集、消费信息的推送等。

单向发送

原理分析

单向(OneWay)发送是一种只负责发送消息而不等待任何响应的方式。生产者将消息发送到 Broker 后(默认超时 5分钟),不关心消息是否成功到达或被持久化,主要依赖 Broker 进行刷盘操作,单向发送通常与异步刷盘结合使用,以提高发送效率。

单向发送主要涉及以下几个步骤:

创建Producer:创建一个Producer对象;

创建消息:创建一个Message对象。

发送消息:调用DefaultMQProducer的sendOneway方法。

如下示例代码为一个完整的单向发送流程:

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.client.producer.SendStatus;

import org.apache.rocketmq.common.message.Message;

public class OneWayProducerTest {

public static void main(String[] args) throws Exception {

// 1、创建 producer,设置组名为 OneWayGroupTest

DefaultMQProducer producer = new DefaultMQProducer("OneWayGroup");

// 2、指定 NameServer的地址,以获取 Broker路由地址

producer.setNamesrvAddr("x.x.x.x:9876");

// 3、启动 producer

producer.start();

// 4、创建消息,并指定Topic,Tag和消息体

Message msg = new Message("OneWayTopic","oneway", "OneWayMessage".getBytes("UTF-8"));

// 5、发送单向消息

producer.sendOneway(msg);

// 6、关闭 Producer

producer.shutdown();

}

}

RocketMQ 的单向发送主要涉及以下几个关键类和方法:

DefaultMQProducer:生产者类,负责发送消息。

MQClientAPIImpl#sendMessage:底层消息发送实现。

NettyRemotingClient#invokeOneway:通过 Netty 实现网络通信。

Broker 端的 SendMessageProcessor:处理发送请求。

源码参考:org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl#sendOneway(Message msg)

优缺点

优点:

非常高效,延迟最低。

适用于对可靠性要求不高的场景。

缺点:

无法确认消息是否成功发送。

可靠性最低,消息可能丢失。

使用场景

适用于对可靠性要求不高的场景,如日志收集、监控数据上报等。

3种方式对比

发送方式

优点

缺点

使用场景

同步发送

可靠性高,简单易用

延迟较高,吞吐量受限

订单系统、金融交易、重要的消息通知等

异步发送

非阻塞,延迟较低

实现复杂度高,可靠性相对降低

实时数据处理、日志采集、消费信息的推送等

单向发送

高效,延迟最低

无法确认消息是否成功发送,可靠性最低

日志收集、监控数据上报等

如何选择?

同步发送

消息发送后会等待服务器的响应,整个过程业务是阻塞等待的,适用于对可靠性要求高的场景,比如 订单系统、金融交易等。

异步发送

消息发送后,不等待服务器响应,而是通过回调函数处理响应,适用于对响应时间要求高的场景,比如实时数据处理、日志采集、消费信息的推送等

单向发送

单向发送只负责发送消息而不等待任何响应的方式,也不需要对发送的状态、结果负责,适用于对可靠性要求不高的场景,比如日志收集、监控数据上报等。

每种发送方式都有其适用的场景和优缺点,具体如何选择,一定需要根据业务需求进行权衡。

总结

本文分析了 RocketMQ 同步发送、异步发送和单向发送三种方式的原理、优缺点以及使用场景,并且分析了每种方式涉及到的核心源码。

通过上文的介绍可以知道同步发送方式可以保证消息发送时不丢,但是性能相对其他两种方式差一些。

RocketMQ 是一款优秀的开源消息中间件,作为 Java程序员,建议多去阅读它的源码,吸收其中比较好的代码思维。

参考资料

rocketmq官网

学习交流

如果你觉得文章有帮助,请帮忙转发给更多的好友,或关注:猿java,持续输出硬核文章。

相关推荐

乌吃什么(食糊,鸟焉能享美食?)
bt365体育投注3

乌吃什么(食糊,鸟焉能享美食?)

⌛ 09-16 👁️ 8590
在手机上怎么开网银?手机开网银
bt365体育投注3

在手机上怎么开网银?手机开网银

⌛ 09-01 👁️ 6113
文质彬彬的司令
英国投注网站365

文质彬彬的司令

⌛ 07-12 👁️ 3409