RocketMQ的最佳实践

代理角色有:异步主节点,同步主节点和从节点. 如果你不能容忍消息丢失,我们建议你部署同步主机并在其上附加从属服务器.如果您对丢失可以容忍,但希望代理始终可用,则可以将ASYNC_MASTER和SLAVE一起部署,如果你只想简单一点,你可能只需要一个没有从节点的异步主机.

刷新磁盘类型

建议使用异步刷新,因为同步刷新时非常昂贵的,会造成太多的新能损失.如果你想要可靠性,我们建议您使用同步主机和丛机.

生产者的最佳实践

对使用者的一些有用的方法.

发送状态

当发送一条消息,你将会得到一个发送状态和一个发送结果,首先我们建设消息的iswaitstoremsgok=true(默认为true), 如果没有异常,我们将始终收到"OK", 以下是每个状态描述列表:

FLUSH_DISK_TIMEOUT:刷新超时时间

如果代理设置了MessageStoreConfig的FlushDiskType=SYNC_FLUSH(默认是ASYNC_FLUSH),并且代理未在messagestoreconfig的syncflushtimeout默认为5秒）内完成磁盘刷新，则将获得此状态.

FLUSH_SLAVE_TIMEOUT:刷新从节点超时

如果代理的角色时同步主机(默认异步主机),如果从机代理没有在MessageStoreConfig的syncFlushTimeout(默认时5秒)时间内异步刷新,则将获得此状态.

SLAVE_NOT_AVAILABLE:从机不可用

如果代理的角色是同步主代理(默认时异步主代理), 但是没有配置从属代理,则将会获得这个状态.

SEND_OK:发送成功

发送"确定"并不意味着它是可靠的,为确保不会丢失任何消息,应该启用同步主机或同步刷新.

Duplication or Missing: 重复或丢失

如果您得到了flush-disk-timeout,flush-slave-timeout,并且代理在此时正好关闭,那么您可以发现您的消息丢失了.这个时候你有两种选择,一个是让它离开,这可能导致这个消息丢失,另一个时重新发送消息,这个可能会导致消息重复.通常,我们建议重新发送并在使用时找到处理重复删除的方法.除非你觉得当一些信息丢失时没有关系.但是请记住,当没有从属主机时,重新发送是无用的,如果发生这种情况,宁应该保留场景并警告集群管理器.

超时

客户端向代理发送请求,并等待响应,但是如果最大等待事件已经过去,并且没有返回响应,则客户端将抛出一个RemotingTimeoutException.默认等待时间时3秒,你可以通过使用send(msg, timeout) 方法来设置超时时间来代替send(msg).注意,我们不建议等待时间设置得太短.因为代理需要一些时间来刷新磁盘或与从属服务器同步.

另外,如果该值超过了syncflushtimeout太多,则效果可能很小,因为代理可能会在超时之前返回FLUSH_SLAVE_TIME 或 FLUSH_SLAVE_TIMEOU

消息体大小

我们建议消息提的大小时不超过512K.

异步发送(Async Sending)

默认的send(msg方法在接收到返回响应消息之前,将会被一直阻塞,因此,如果宁关心性能,我们建议您使用send(msg, callback), 它将以异步的方式进行工作.

生产组(Producer Group)

通常,生产者组是没有影响的.但是如果你加入了一个事物,你就应该注意,默认情况下,在同一个JVM中只能创建一个具有相同生产组的生产这,这通常已经足够了.

线程安全

生产者是线程安全的.您可以业务解决方案中使用它.

性能

如果你希望在一个JVM中有多个生产者进行大数据处理.我们建议:

• 使用几个生产者异步发送(3-5个就足够了)
• 为每个生产者设置实例名称

消费的最佳实践

为用户提供一些有用的提示

消费组和订阅

首先,你需要注意的是,不用的消费组可以独立的消费同一个主题,并且每个消费组都有自己的消费补偿.请确保同一组中的每个消费者订阅相同的主题.

消息监听器

Orderly

消费者将锁定每个消息队列,以确保按顺序逐个消费它.这个将导致性能损失,但当你关系消息的顺序时,它是有用的.不建议抛出异常,你可以返回ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT.

Concurrently

顾名思义,使用者将并发消费.建议使用它以获得良好的性能,不建议抛出异常,你可以使用ConsumeConcurrentlyStatus.RECONSUME_LATER 来进行替代.

Consume Status

对MessageListenerConcurrently来说,你可以返回 RECONSUME_LATER 告诉消费者你现在不能马上消费它,想稍等一会儿再消费它,然后你可以继续消费其他消息,对MessageListenerOrderly来说,因为你关心的是顺序,你不能跳过某一条消息,但是你可以返回 SUSPEND_CURRENT_QUEUE_A_MOMENT (暂停当前队列)来告诉消费者等待一会儿

Blocking

不建议阻塞监听器,因为它会阻塞线程池,最终可能会导致消费进程的停止.

线程数量

消费者使用 ThreadPoolExecutor 来处理内部消费,因此您可以通过 setConsumeThreadMin 或者 setConsumeThreadMax 来改变最大消费线程数和最大消费线程数.

从哪里开始消费

当一个新的消费组建立时,它将需要决定是否需要消费已经存在与代理中的历史消息.

CONSUME_FROM_LAST_OFFSET: 这个配置将会忽略历史消息,并消费之后产生的任何内容.

CONSUME_FROM_FIRST_OFFSET: 这个配置将会消费代理中存在的所有消息.

CONSUME_FROM_TIMESTAMP: 这个配置将消费指定时间戳之后再生成消息.

重复

许多情况可能导致重复,例如:

• 生产者重复发送(例如: 在FLUSH_SLAVE_TIMEOUT的情况下)
• 消费者关闭,某些补偿机制未及时更新到代理

因此,如果应用程序不能容忍重复,你可能需要做一些外部工作来处理这一问题,例如: 通过检查数据库的主键.

名称服务的最佳实践

在ApacheRocketMQ中,名称服务器被设计为协调分布式系统的每个组件,协调主要通过管理主题路由信息来实现.

管理主要由两部分构成:

• 代理定期更新保存在每个名称服务器中的元数据
• 名称服务器为客户端服务,包括生产者,消费者和命令行客户机提供最新的路由信息

因此,在启动代理和客户端之前,我们需要告诉他们如果通过名称服务器提供名称服务器地址列表来访问名称服务器.在Apache RocketMQ中可以通过四种方式实现.

编程的方式

Java配置项

名称服务器地址列表可以通过在启动之前指定后继 Java 选项rocketmq.namesrv.addr来提供给应用程序.