标签归档:rocketmq

RocketMq知识点理解

RocketMq关键点理解

各个角色间的关系:

RocketMq中每个Broker(master和slave)与Name Server集群中的所有节点建立长连接,定时注册Topic信息到所有Name Server。
Name Server之间不会有任何信息交互,各自独立。
producer和consumer随机从一个name server即可获得全部topic的路由信息。
producer根据得到的路由信息,同master建立长连接。
consumer根据路由信息,同master个salve都建立长连接。然后根据设置的订阅规则,选择从master或者slave订阅消息。

Broker:

分为master和slave两个角色。master提供读写读物,salve只提供读服务。
为了保证可用性,需要部署多套broker,每套broker至少有1个master和1个以上的salve。
在同一套broker中,master和salve都是同样的brokerName,master的brokerId是0,salve的brokerId必须是非0的。

同步刷盘和异步刷盘。

同步刷盘是说,broker在收到每个消息后,都是先要保存到硬盘上,然后再给producer确认。异步刷盘就是先回复确认,然后批量保存到硬盘上。异步刷盘有更好的性能,当然也有更大的丢失消息的风险。

同步复制和异步复制。

是说在master和salve之间复制消息的方式。同步是说在salve也存储了消息后再答复producer。
异步复制是先答复producer,再去向salve复制。
通过同步复制技术可以完全避免单点,同步复制势必会影响性能,适合对消息可靠性要求极高的场合,例如与Money相关的应用。RocketMQ从3.0版本开始支持同步双写。

MQPullConsumer和MQPushConsumer的区别

consumer被分为2类:MQPullConsumer和MQPushConsumer,其实本质都是拉模式(pull),即consumer轮询从broker拉取消息。
区别是:
push方式里,consumer把轮询过程封装了,并注册MessageListener监听器,取到消息后,唤醒MessageListener的consumeMessage()来消费,对用户而言,感觉消息是被推送(push)过来的。
pull方式里,取消息的过程需要用户自己写,首先通过打算消费的Topic拿到MessageQueue的集合,遍历MessageQueue集合,然后针对每个MessageQueue批量取消息,一次取完后,记录该队列下一次要取的开始offset,直到取完了,再换另一个MessageQueue。

RocketMQ使用长轮询Pull方式,可保证消息非常实时,消息实时性不低于Push的理解

数据交互有两种模式:Push(推模式)、Pull(拉模式)。
推模式指的是客户端与服务端建立好网络长连接,服务方有相关数据,直接通过长连接通道推送到客户端。其优点是及时,一旦有数据变更,客户端立马能感知到;另外对客户端来说逻辑简单,不需要关心有无数据这些逻辑处理。缺点是不知道客户端的数据消费能力,可能导致数据积压在客户端,来不及处理。
拉模式指的是客户端主动向服务端发出请求,拉取相关数据。其优点是此过程由客户端发起请求,故不存在推模式中数据积压的问题。缺点是可能不够及时,对客户端来说需要考虑数据拉取相关逻辑,何时去拉,拉的频率怎么控制等等。

长轮询:
轮询是说,每隔一定时间,客户端想服务端发起一次请求,服务端有数据就返回数据,没有数据就返回空,然后关闭请求。
长轮询,不同之处是,服务端如果此时没有数据,保持连接。等到有数据返回(相当于一种push),或者超时返回。
所以长轮询Pull的好处就是可以减少无效请求,保证消息的实时性,又不会造成客户端积压。

其他

  1. 对于一个消息中间件来说,持久化部分的性能直接决定了整个消息中间件的性能。RocketMQ充分利用Linux文件系统内存cache来提高性能
  2. RocketMq Broker的buffer不会满。原因是RocketMQ没有内存Buffer概念,RocketMQ的队列都是持久化磁盘,数据定期清除。这是RocketMq和其他消息中间件的重要区别。对RocketMQ来说的内存Buffer抽象成一个无限长度的队列,不管有多少数据进来都能装得下,这个无限是有前提的,Broker会定期删除过期的数据。
  3. 消息堆积。消息堆积的能力是评价一个消息中间件的重要方面。因为使用消息中间件有一部分功能是为了为后端系统挡住数据洪峰。在产生消息堆积时,消息中间件对外的服务能力至关重要。
    因为RocketMq的消息都是持久化硬盘的,当消息不能在内存Cache命中时,要不可避免的访问磁盘,会产生大量读IO,读IO的吞吐量直接决定了消息堆积后的访问能力。
  4. 分布式事务。分布式事务涉及到两阶段提交。分为预提交阶段和commit阶段。在commit阶段需要回去改消息的状态。RocketMq在这里没有使用KV存储来做。而是在commit阶段会拿到消息的offset,然后直接去找消息,修改其状态。这样的好处是设计更简单,速度更快。缺点是会产生过多的数据脏页。
  5. producer只与master建立连接,consumer同master和slave都建立连接,向谁订阅可以配置。
  6. Rocketmq的消息的存储是由consume queue和 commitLog 配合完成的
    consume queue 消息的逻辑队列,相当于字典的目录用来指定消息在消息的真正的物理文件commitLog上的位置, 每个topic下的每个queue都有一个对应的consumequeue文件。
    默认文件地址:${user.home} \store\consumequeue${topicName}${queueId}${fileName}
    修改方法:配置文件的
    storePathRootDir=/home/haieradmin/mqstore/rocketmqstore
    storePathCommitLog=/home/haieradmin/mqstore/rocketmqstore/commitlog
    这两个参数。

 

说到高性能消息中间件,第一个想到的肯定是 LinkedIn 开源的 Kafka ,虽然最初 Kafka 是为日志传输而生,但也非常适合互联网公司消息服务的应用场景,他们不要求数据实时的强一致性(事务),更多是希望达到数据的最终一致性。 RocketMQ 是 MetaQ 的 3.0 版本,而 MetaQ 最初的设计又参考了 Kafka 。MetaQ 1.x 和 MetaQ 2.x 是依赖 ZooKeeper 的,但 RocketMQ (即 MetaQ 3.x )却去掉了 ZooKeeper 依赖,转而采用自己的 NameServer (专门为Rocket设计的轻量级名称服务)。接下来看一下NameSever在RocketMQ中的作用想必大家就清楚了。

一:rocketmq为什么使用nameserver?
RocketMQ 的 Broker 有三种集群部署方式: 1. 单台 Master 部署; 2. 多台 Master部署; 3. 多 Master 多 Slave 部署;采用第 3 种部署方式时, Master 和 Slave 可以采用同步复制和异步复制两种方式。下图是第 3 种部署方式的简单图:

图1.多master多slave部署

下面解读一下它们之间的关系:
1.1.NameServer互相独立,彼此没有通信关系,单台NameServer挂掉,不影响其他NameServer。NameServer不去连接别的机器,不会主动推消息。

1.2.单个broker(Master、Slave)与所有NameServer进行定时注册,以便告知NameServer自己还活着。Broker每隔30秒向所有NameServer发送心跳,心跳包含了自身的topic配置信息。NameServer每隔10秒,扫描所有还存活的broker连接,如果某个连接的最后更新时间与当前时间差值超过2分钟,则断开此连接。此外,NameServer也会断开此broker下所有与slave的连接。同时更新topic与队列的对应关系,但不会通知生产者和消费者。
Broker slave 同步或者异步从Broker master 上拷贝数据。

1.3.consumer随机与一个NameServer建立长连接,如果该NameServer断开,则从NameServer列表中查找下一个进行连接。
consumer主要从NameServer中根据topic查询broker的地址,查到就会缓存到客户端,并向提供topic服务的master、slave建立长连接,且定时向master、slave发送心跳。如果broker宕机,则NameServer会将其剔除,而consumer端的定时任务MQClientInstance.this.updateTopicRouteInfoFromNameServer每30秒执行一次,会将topic对应的broker地址拉取下来,此地址已经为slave地址了,故此时consumer会从slave上消费。 消费者与master和slave都建有连接,在不同场景有不同的消费规则。

1.4.Producer随机与一个NameServer建立长连接,每隔30秒(此处时间可配置)从NameServer获取topic的最新队列情况,这就表示如果某个broker master宕机,producer最多30秒才能感知,在这个期间,发往该broker master的消息将会失败。Producer会向提供topic服务的master建立长连接,且定时向master发送心跳。生产者与所有的master连接,但不能向slave写入。
客户端是先从NameServer寻址的,得到可用Broker的IP和端口信息,然后自己去连接broker。
(ps:我们在设计系统的时候,有一些全局使用的公用信息,可以单独独立一个模块进行专门的管理;各个子模块只需要定时向该模块更新信息即可。)

综上所述,我们可以得出NameServer在RocketMQ中所扮演的角色:1.NameServer 用来保存活跃的 broker 列表,包括 Master 和 Slave 。
2.NameServer 用来保存所有 topic 和该 topic 所有队列的列表。3.NameServer 用来保存所有 broker 的 Filter 列表。
对于RocketMQ 来说,topic 的数据在每个 Master 上是对等的,没有哪个 Master 上有 topic 上的全部数据,所以对于zookeeper的Master 选举功能在Rocket中使用不到。
Broker与slave配对是通过指定相同的brokerName参数来配对,master的brokerId必须为0,slave的brokerId必须大于0,此外一个master下可以挂多个slave,同一个master下的多个slave通过指定不同的brokerId来区分。

二:核心模块
2.1.注册broker

2.2 存放消息
Producer在存放消息时,首先会使用GET_ROUTEINTO_BY_TOPIC获取route信息TopicRouteData。
Producer Group–Producer实例的集合。
Producer实例可以是多机器、但机器多进程、单进程中的多对象。Producer可以发送多个Topic。处理分布式事务时,也需要Producer集群提高可靠性。

2.3 读取消息
Consumer在读取消息时,也会先使用GET_ROUTEINTO_BY_TOPIC获取route信息TopicRouteData。
Consumer Group–Consumer实例的集合。
Consumer 实例可以是多机器、但机器多进程、单进程中的多对象。同一个Group中的实例,在集群模式下,以均摊的方式消费;在广播模式下,每个实例都全部消费。

2.3.1.Push Consumer
Consumer的一种,应用通常向Consumer对象注册一个Listener接口,一旦收到消息,Consumer对象立刻回调Listener接口方法。所以,所谓Push指的是客户端内部的回调机制,并不是与服务端之间的机制。

2.3.2.Pull Consumer
Consumer的一种,应用通常主动调用Consumer从服务端拉消息,consumer 向 broker 发送拉消息请求, PullMessageService 服务通过一个线程将阻塞队列 LinkedBlockingQueue 中的 PullRequest 到 broker 拉取消息,长轮询向broker拉取消息是批量拉取的,默认设置批量的值为pullBatchSize = 32, 可配置。主动权由应用控制,这用的就是短轮询方式了,在不同情况下,与长轮询各有优点。

三:消息
*Message priority
处于一个队列,可用整数来描述每条消息的优先级,在投递前先按优先级排好序,然后令优先级高的消息先投递。RocketMQ没有特意支持消息优先级,它的所有的消息都是持久化,如果按优先级排序,开销会非常大,可以通过变通方式实现类似功能,即不同优先级各对应一个队列,发送消息时根据优先级发送到与之对应的队列中。

*Message order
rocketmq的顺序消息需要满足2点:
1.Producer端保证发送消息有序,且发送到同一个队列。
2.consumer端保证消费同一个队列。
顺序消费的发送的原理:
SendResult sendResult = producer.send(msg, new MessageQueueSelector()
{
@Override
public MessageQueue select(List mqs,Message msg, Object arg)
{
Integer id = (Integer) arg;
int index = id % mqs.size(); // 如果队列数不变,同一个订单号取到的队列是同一个
return mqs.get(index);
}
}, “10001”); // orderID “10001”是传递给回调方法的自定义数据
}
//List mqs 就是从namesrv 获取的所有队列

那么在集群消费时它是如何做到消息的有序性?
1.ConsumeMessageOrderlyService类的start()方法,如果是集群消费,则启动定时任务,定时向broker发送批量锁住当前正在消费的队列集合的消息,具体是consumer端拿到正在消费的队列集合,发送锁住队列的消息至broker,broker端返回锁住成功的队列集合。 consumer收到后,设置是否锁住标志位。

2.ConsumeMessageOrderlyService.ConsumeRequest的run方法是消费消息,这里还有个MessageQueueLock messageQueueLock,维护当前consumer端的本地队列锁。保证当前只有一个线程能够进行消费。

3.拉到消息存入ProcessQueue,然后判断,本地是否获得锁,全局队列是否被锁住,然后从ProcessQueue里取出消息,用MessageListenerOrderly进行消费。 拉到消息后调用ProcessQueue.putMessage(final List msgs) 存入,具体是存入TreeMap msgTreeMap。
然后是调用ProcessQueue.takeMessags(final int batchSize)消费,具体是把msgTreeMap里消费过的消息,转移到TreeMap msgTreeMapTemp。

4.本地消费的事务控制,ConsumeOrderlyStatus.SUCCESS(提交),ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT(挂起一会再消费),在此之前还有一个变量ConsumeOrderlyContext context的setAutoCommit()是否自动提交。

当SUSPEND_CURRENT_QUEUE_A_MOMENT时,autoCommit设置为true或者false没有区别,本质跟消费相反,把消息从msgTreeMapTemp转移回msgTreeMap,等待下次消费。
当SUCCESS时,autoCommit设置为true时比设置为false多做了2个动作,consumeRequest.getProcessQueue().commit()和this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset(consumeRequest.getMessageQueue(), commitOffset, false);
ProcessQueue.commit() :本质是删除msgTreeMapTemp里的消息,msgTreeMapTemp里的消息在上面消费时从msgTreeMap转移过来的。
this.defaultMQPushConsumerImpl.getOffsetStore().updateOffset() :本质是把拉消息的偏移量更新到本地,然后定时更新到broker。

Listener 接口方法:
MessageListenerOrderly ->是有序的
MessageListenerConcurrently ->是无序的

*Message Filter
Broker端过滤-RocketMQ支持根据msgId、key、Header、body进行过滤。根据Message Key查询仅仅返回符合条件的最近64条数据,所以key值尽量唯一,并且有业务区分度。

*发送消息负载均衡
使用Roundbin方式,轮询发送消息,每个队列接收平均的消息量。也可以自定义选择发往哪个队列。

 

RocketMQ初探一:NameServer的作用

第一次真正接触Java消息服务是在2013年底,当时是给中国移动做统一支付平台,当时用的就是著名的Apache ActiveMQ,当时觉得很有趣,一个服务队列竟然可以玩出这么多花样来。当时为了尽快的入门,还把《Java Message Service》给看了一遍,这对于初学者的我收获颇多。我说知道的完全实现JMS规范的MOM有ActiveMQ/Apollo和HornetQ,都是采用Java实现。JMS中说明了Java消息服务的两种消息传送模型,即P2P(点对点)和Pub/Sub(发布订阅),在约定了一些消息服务特性的同时,并提供了一套接口API,是否实现了该API,标志着MOM是否支持JMS规范,JMS规范中定义了消息服务诸多特性,这些特性和他所面对的企业级服务场景相关,当然,这也严重限制了消息服务的吞吐量,完全实现JMS规范的MOM,性能总不会太高,而且JMS规范中没有涉及消息服务的分布式特性,导致大多数实现JMS规范的MOM分布式部署功能比较弱,只适合集群部署。

 

说到高性能消息中间件,第一个想到的肯定是LinkedIn开源的Kafka,虽然最初Kafka是为日志传输而生,但也非常适合互联网公司消息服务的应用场景,他们不要求数据实时的强一致性(事务),更多是希望达到数据的最终一致性。RocketMQ是MetaQ的3.0版本,而MetaQ最初的设计又参考了Kafka。最初的MetaQ 1.x版本由阿里的原作者庄晓丹开发,后面的MetaQ 2.x版本才进行了开源,这里需要注意一点的事,MetaQ 1.x和MetaQ 2.x是依赖ZooKeeper的,但RocketMQ(即MetaQ 3.x)却去掉了ZooKeeper依赖,转而采用自己的NameServer。

 

ZooKeeper是著名的分布式协作框架,提供了Master选举、分布式锁、数据的发布和订阅等诸多功能,为什么RocketMQ没有选择ZooKeeper,而是自己开发了NameServer,我们来具体看看NameServer在RocketMQ集群中的作用就明了了。

 

RocketMQ的Broker有三种集群部署方式:1.单台Master部署;2.多台Master部署;3.多Master多Slave部署;采用第3种部署方式时,Master和Slave可以采用同步复制和异步复制两种方式。下图是第3种部署方式的简单图:

 


图虽然是网上找的,但也足以说明问题,当采用多Master方式时,Master与Master之间是不需要知道彼此的,这样的设计直接降低了Broker实现的复查性,你可以试想,如果Master与Master之间需要知道彼此的存在,这会需要在Master之中维护一个网络的Master列表,而且必然设计到Master发现和活跃Master数量变更等诸多状态更新问题,所以最简单也最可靠的做法就是Master只做好自己的事情(比如和Slave进行数据同步)即可,这样,在分布式环境中,某台Master宕机或上线,不会对其他Master造成任何影响。

 

那么怎么才能知道网络中有多少台Master和Slave呢?你会很自然想到用ZooKeeper,每个活跃的Master或Slave都去约定的ZooKeeper节点下注册一个状态节点,但RocketMQ没有使用ZooKeeper,所以这件事就交给了NameServer来做了(看上图)。

 

结论一:NameServer用来保存活跃的broker列表,包括Master和Slave。

当然,这个结论百度一查就知道,我们移步到rocketmq-namesrv模块中最重要的一个类:RouteInfoManager,它的主要属性如下:

 

private final ReadWriteLock lock = new ReentrantReadWriteLock();

private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;

private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;

private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;

private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;

private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

 

每个属性通过名字就能清楚的知道是什么意思,之所以能用非线程安全的HashMap,是因为有读写锁lock来对HashMap的修改做保护。我们注意到保存broker的Map有两个,即brokerAddrTable用来保存所有的broker列表和brokerLiveTable用来保存当前活跃的broker列表,而BrokerData用来保存broker的主要新增,而BrokerLiveInfo只用来保存上次更新(心跳)时间,我们可以直接看看RouteInfoManager中扫描非活跃broker的方法:

 

// Broker Channel两分钟过期

private final static long BrokerChannelExpiredTime = 1000 * 60 * 2;

public void scanNotActiveBroker() {

Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();

while (it.hasNext()) {

Entry<String, BrokerLiveInfo> next = it.next();

long last = next.getValue().getLastUpdateTimestamp();

if ((last + BrokerChannelExpiredTime) < System.currentTimeMillis()) {

RemotingUtil.closeChannel(next.getValue().getChannel());

it.remove();

log.warn("The broker channel expired, {} {}ms", next.getKey(), BrokerChannelExpiredTime);

this.onChannelDestroy(next.getKey(), next.getValue().getChannel());

}

}

}

 

可以看出,如果两分钟内都没收到一个broker的心跳数据,则直接将其从brokerLiveTable中移除,注意,这还会导致该broker从brokerAddrTable被删除,当然,如果该broker是Master,则它的所有Slave的broker都将被删除。具体细节可以参看RouteInfoManager的onChannelDestroy方法。

 

结论二:NameServer用来保存所有topic和该topic所有队列的列表。

我们注意到,topicQueueTable的value是QueueData的List,我们看看QueueData中的属性:

 

private String brokerName;  // broker的名称

private int readQueueNums;  // 读队列数量

private int writeQueueNums; // 写队列数量

private int perm;           // 读写权限

private int topicSynFlag;   // 同步复制还是异步复制标记

 

所以,你几乎可以在NameServer这里知道topic相关的所有信息,包括topic有哪些队列,这些队列在那些broker上等。

 

结论三:NameServer用来保存所有broker的Filter列表。

关于这一点,讨论broker的时候再细说。

 

DefaultRequestProcessor是NameServer的默认请求处理器,他处理了定义在rocketmq-common模块中RequestCode定义的部分请求,比如注册broker、注销broker、获取topic路由、删除topic、获取broker的topic权限、获取NameServer的所有topic等。

 

在源代码中,NettyServerConfig类记录NameServer中的一些默认参数,比如端口、服务端线程数等,列出如下:

private int listenPort = 8888;

private int serverWorkerThreads = 8;

private int serverCallbackExecutorThreads = 0;

private int serverSelectorThreads = 3;

private int serverOnewaySemaphoreValue = 256;

private int serverAsyncSemaphoreValue = 64;

private int serverChannelMaxIdleTimeSeconds = 120;

 

这些都可以通过启动时指定配置文件来进行覆盖修改,具体可以参考NameServer的启动类NamesrvStartup的实现(没想到Apache还有对命令行提供支持的commons-cls的包)。

 

 

现在我们再回过头来看看RocketMQ为什么不使用ZooKeeper?ZooKeeper可以提供Master选举功能,比如Kafka用来给每个分区选一个broker作为leader,但对于RocketMQ来说,topic的数据在每个Master上是对等的,没有哪个Master上有topic上的全部数据,所以这里选举leader没有意义;RockeqMQ集群中,需要有构件来处理一些通用数据,比如broker列表,broker刷新时间,虽然ZooKeeper也能存放数据,并有一致性保证,但处理数据之间的一些逻辑关系却比较麻烦,而且数据的逻辑解析操作得交给ZooKeeper客户端来做,如果有多种角色的客户端存在,自己解析多级数据确实是个麻烦事情;既然RocketMQ集群中没有用到ZooKeeper的一些重量级的功能,只是使用ZooKeeper的数据一致性和发布订阅的话,与其依赖重量级的ZooKeeper,还不如写个轻量级的NameServer,NameServer也可以集群部署,NameServer与NameServer之间无任何信息同步,只有一千多行代码的NameServer稳定性肯定高于ZooKeeper,占用的系统资源也可以忽略不计,何乐而不为?当然,这些只是本人的一点理解,具体原因当然得RocketMQ设计和开发者来说。

来源:http://manzhizhen.iteye.com/blog/2317354

 

分布式开放消息系统(RocketMQ)的原理与实践

分布式消息系统作为实现分布式系统可扩展、可伸缩性的关键组件,需要具有高吞吐量、高可用等特点。而谈到消息系统的设计,就回避不了两个问题:

  1. 消息的顺序问题
  2. 消息的重复问题

RocketMQ作为阿里开源的一款高性能、高吞吐量的消息中间件,它是怎样来解决这两个问题的?RocketMQ 有哪些关键特性?其实现原理是怎样的?

关键特性以及其实现原理

一、顺序消息

消息有序指的是可以按照消息的发送顺序来消费。例如:一笔订单产生了 3 条消息,分别是订单创建、订单付款、订单完成。消费时,要按照顺序依次消费才有意义。与此同时多笔订单之间又是可以并行消费的。首先来看如下示例:

假如生产者产生了2条消息:M1、M2,要保证这两条消息的顺序,应该怎样做?你脑中想到的可能是这样:

你可能会采用这种方式保证消息顺序

假定M1发送到S1,M2发送到S2,如果要保证M1先于M2被消费,那么需要M1到达消费端被消费后,通知S2,然后S2再将M2发送到消费端。

这个模型存在的问题是,如果M1和M2分别发送到两台Server上,就不能保证M1先达到MQ集群,也不能保证M1被先消费。换个角度看,如果M2先于M1达到MQ集群,甚至M2被消费后,M1才达到消费端,这时消息也就乱序了,说明以上模型是不能保证消息的顺序的。如何才能在MQ集群保证消息的顺序?一种简单的方式就是将M1、M2发送到同一个Server上:

保证消息顺序,你改进后的方法

这样可以保证M1先于M2到达MQServer(生产者等待M1发送成功后再发送M2),根据先达到先被消费的原则,M1会先于M2被消费,这样就保证了消息的顺序。

这个模型也仅仅是理论上可以保证消息的顺序,在实际场景中可能会遇到下面的问题:

网络延迟问题

只要将消息从一台服务器发往另一台服务器,就会存在网络延迟问题。如上图所示,如果发送M1耗时大于发送M2的耗时,那么M2就仍将被先消费,仍然不能保证消息的顺序。即使M1和M2同时到达消费端,由于不清楚消费端1和消费端2的负载情况,仍然有可能出现M2先于M1被消费的情况。

那如何解决这个问题?将M1和M2发往同一个消费者,且发送M1后,需要消费端响应成功后才能发送M2。

聪明的你可能已经想到另外的问题:如果M1被发送到消费端后,消费端1没有响应,那是继续发送M2呢,还是重新发送M1?一般为了保证消息一定被消费,肯定会选择重发M1到另外一个消费端2,就如下图所示。

保证消息顺序的正确姿势

这样的模型就严格保证消息的顺序,细心的你仍然会发现问题,消费端1没有响应Server时有两种情况,一种是M1确实没有到达(数据在网络传送中丢失),另外一种消费端已经消费M1且已经发送响应消息,只是MQ Server端没有收到。如果是第二种情况,重发M1,就会造成M1被重复消费。也就引入了我们要说的第二个问题,消息重复问题,这个后文会详细讲解。

回过头来看消息顺序问题,严格的顺序消息非常容易理解,也可以通过文中所描述的方式来简单处理。总结起来,要实现严格的顺序消息,简单且可行的办法就是:

保证生产者 - MQServer - 消费者是一对一对一的关系

这样的设计虽然简单易行,但也会存在一些很严重的问题,比如:

  1. 并行度就会成为消息系统的瓶颈(吞吐量不够)
  2. 更多的异常处理,比如:只要消费端出现问题,就会导致整个处理流程阻塞,我们不得不花费更多的精力来解决阻塞的问题。

但我们的最终目标是要集群的高容错性和高吞吐量。这似乎是一对不可调和的矛盾,那么阿里是如何解决的?

世界上解决一个计算机问题最简单的方法:“恰好”不需要解决它!—— 沈询

有些问题,看起来很重要,但实际上我们可以通过合理的设计或者将问题分解来规避。如果硬要把时间花在解决问题本身,实际上不仅效率低下,而且也是一种浪费。从这个角度来看消息的顺序问题,我们可以得出两个结论:

  1. 不关注乱序的应用实际大量存在
  2. 队列无序并不意味着消息无序

所以从业务层面来保证消息的顺序而不仅仅是依赖于消息系统,是不是我们应该寻求的一种更合理的方式?

最后我们从源码角度分析RocketMQ怎么实现发送顺序消息。

RocketMQ通过轮询所有队列的方式来确定消息被发送到哪一个队列(负载均衡策略)。比如下面的示例中,订单号相同的消息会被先后发送到同一个队列中:

// RocketMQ通过MessageQueueSelector中实现的算法来确定消息发送到哪一个队列上
// RocketMQ默认提供了两种MessageQueueSelector实现:随机/Hash
// 当然你可以根据业务实现自己的MessageQueueSelector来决定消息按照何种策略发送到消息队列中
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Integer id = (Integer) arg;
        int index = id % mqs.size();
        return mqs.get(index);
    }
}, orderId);

在获取到路由信息以后,会根据MessageQueueSelector实现的算法来选择一个队列,同一个OrderId获取到的肯定是同一个队列。

private SendResult send()  {
    // 获取topic路由信息
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    if (topicPublishInfo != null && topicPublishInfo.ok()) {
        MessageQueue mq = null;
        // 根据我们的算法,选择一个发送队列
        // 这里的arg = orderId
        mq = selector.select(topicPublishInfo.getMessageQueueList(), msg, arg);
        if (mq != null) {
            return this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
        }
    }
}
二、消息重复

上面在解决消息顺序问题时,引入了一个新的问题,就是消息重复。那么RocketMQ是怎样解决消息重复的问题呢?还是“恰好”不解决。

造成消息重复的根本原因是:网络不可达。只要通过网络交换数据,就无法避免这个问题。所以解决这个问题的办法就是绕过这个问题。那么问题就变成了:如果消费端收到两条一样的消息,应该怎样处理?

  1. 消费端处理消息的业务逻辑保持幂等性
  2. 保证每条消息都有唯一编号且保证消息处理成功与去重表的日志同时出现

第1条很好理解,只要保持幂等性,不管来多少条重复消息,最后处理的结果都一样。第2条原理就是利用一张日志表来记录已经处理成功的消息的ID,如果新到的消息ID已经在日志表中,那么就不再处理这条消息。

第1条解决方案,很明显应该在消费端实现,不属于消息系统要实现的功能。第2条可以消息系统实现,也可以业务端实现。正常情况下出现重复消息的概率其实很小,如果由消息系统来实现的话,肯定会对消息系统的吞吐量和高可用有影响,所以最好还是由业务端自己处理消息重复的问题,这也是RocketMQ不解决消息重复的问题的原因。

RocketMQ不保证消息不重复,如果你的业务需要保证严格的不重复消息,需要你自己在业务端去重。

三、事务消息

RocketMQ除了支持普通消息,顺序消息,另外还支持事务消息。首先讨论一下什么是事务消息以及支持事务消息的必要性。我们以一个转帐的场景为例来说明这个问题:Bob向Smith转账100块。

在单机环境下,执行事务的情况,大概是下面这个样子:

单机环境下转账事务示意图

当用户增长到一定程度,Bob和Smith的账户及余额信息已经不在同一台服务器上了,那么上面的流程就变成了这样:

集群环境下转账事务示意图

这时候你会发现,同样是一个转账的业务,在集群环境下,耗时居然成倍的增长,这显然是不能够接受的。那如何来规避这个问题?

大事务 = 小事务 + 异步

将大事务拆分成多个小事务异步执行。这样基本上能够将跨机事务的执行效率优化到与单机一致。转账的事务就可以分解成如下两个小事务:

小事务+异步消息

图中执行本地事务(Bob账户扣款)和发送异步消息应该保证同时成功或者同时失败,也就是扣款成功了,发送消息一定要成功,如果扣款失败了,就不能再发送消息。那问题是:我们是先扣款还是先发送消息呢?

首先看下先发送消息的情况,大致的示意图如下:

事务消息:先发送消息

存在的问题是:如果消息发送成功,但是扣款失败,消费端就会消费此消息,进而向Smith账户加钱。

先发消息不行,那就先扣款吧,大致的示意图如下:

事务消息-先扣款

存在的问题跟上面类似:如果扣款成功,发送消息失败,就会出现Bob扣钱了,但是Smith账户未加钱。

可能大家会有很多的方法来解决这个问题,比如:直接将发消息放到Bob扣款的事务中去,如果发送失败,抛出异常,事务回滚。这样的处理方式也符合“恰好”不需要解决的原则。

这里需要说明一下:如果使用Spring来管理事物的话,大可以将发送消息的逻辑放到本地事物中去,发送消息失败抛出异常,Spring捕捉到异常后就会回滚此事物,以此来保证本地事物与发送消息的原子性。

RocketMQ支持事务消息,下面来看看RocketMQ是怎样来实现的。

RocketMQ实现发送事务消息

RocketMQ第一阶段发送Prepared消息时,会拿到消息的地址,第二阶段执行本地事物,第三阶段通过第一阶段拿到的地址去访问消息,并修改消息的状态。

细心的你可能又发现问题了,如果确认消息发送失败了怎么办?RocketMQ会定期扫描消息集群中的事物消息,如果发现了Prepared消息,它会向消息发送端(生产者)确认,Bob的钱到底是减了还是没减呢?如果减了是回滚还是继续发送确认消息呢?RocketMQ会根据发送端设置的策略来决定是回滚还是继续发送确认消息。这样就保证了消息发送与本地事务同时成功或同时失败。

那我们来看下RocketMQ源码,是如何处理事务消息的。客户端发送事务消息的部分(完整代码请查看:rocketmq-example工程下的com.alibaba.rocketmq.example.transaction.TransactionProducer):

// =============================发送事务消息的一系列准备工作========================================
// 未决事务,MQ服务器回查客户端
// 也就是上文所说的,当RocketMQ发现`Prepared消息`时,会根据这个Listener实现的策略来决断事务
TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
// 构造事务消息的生产者
TransactionMQProducer producer = new TransactionMQProducer("groupName");
// 设置事务决断处理类
producer.setTransactionCheckListener(transactionCheckListener);
// 本地事务的处理逻辑,相当于示例中检查Bob账户并扣钱的逻辑
TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
producer.start()
// 构造MSG,省略构造参数
Message msg = new Message(......);
// 发送消息
SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
producer.shutdown();

接着查看sendMessageInTransaction方法的源码,总共分为3个阶段:发送Prepared消息、执行本地事务、发送确认消息。

//  ================================事务消息的发送过程=============================================
public TransactionSendResult sendMessageInTransaction(.....)  {
    // 逻辑代码,非实际代码
    // 1.发送消息
    sendResult = this.send(msg);
    // sendResult.getSendStatus() == SEND_OK
    // 2.如果消息发送成功,处理与消息关联的本地事务单元
    LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
    // 3.结束事务
    this.endTransaction(sendResult, localTransactionState, localException);
}

endTransaction方法会将请求发往broker(mq server)去更新事务消息的最终状态:

  1. 根据sendResult找到Prepared消息sendResult包含事务消息的ID
  2. 根据localTransaction更新消息的最终状态

如果endTransaction方法执行失败,数据没有发送到broker,导致事务消息的 状态更新失败,broker会有回查线程定时(默认1分钟)扫描每个存储事务状态的表格文件,如果是已经提交或者回滚的消息直接跳过,如果是prepared状态则会向Producer发起CheckTransaction请求,Producer会调用DefaultMQProducerImpl.checkTransactionState()方法来处理broker的定时回调请求,而checkTransactionState会调用我们的事务设置的决断方法来决定是回滚事务还是继续执行,最后调用endTransactionOnewaybroker来更新消息的最终状态。

再回到转账的例子,如果Bob的账户的余额已经减少,且消息已经发送成功,Smith端开始消费这条消息,这个时候就会出现消费失败和消费超时两个问题,解决超时问题的思路就是一直重试,直到消费端消费消息成功,整个过程中有可能会出现消息重复的问题,按照前面的思路解决即可。

消费事务消息

这样基本上可以解决消费端超时问题,但是如果消费失败怎么办?阿里提供给我们的解决方法是:人工解决。大家可以考虑一下,按照事务的流程,因为某种原因Smith加款失败,那么需要回滚整个流程。如果消息系统要实现这个回滚流程的话,系统复杂度将大大提升,且很容易出现Bug,估计出现Bug的概率会比消费失败的概率大很多。这也是RocketMQ目前暂时没有解决这个问题的原因,在设计实现消息系统时,我们需要衡量是否值得花这么大的代价来解决这样一个出现概率非常小的问题,这也是大家在解决疑难问题时需要多多思考的地方。

20160321补充:在3.2.6版本中移除了事务消息的实现,所以此版本不支持事务消息,具体情况请参考rocketmq的issues:

https://github.com/alibaba/RocketMQ/issues/65

https://github.com/alibaba/RocketMQ/issues/138

https://github.com/alibaba/RocketMQ/issues/156

四、Producer如何发送消息

Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡,如下图所示:

producer发送消息负载均衡

首先分析一下RocketMQ的客户端发送消息的源码:

// 构造Producer
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
// 初始化Producer,整个应用生命周期内,只需要初始化1次
producer.start();
// 构造Message
Message msg = new Message("TopicTest1",// topic
                        "TagA",// tag:给消息打标签,用于区分一类消息,可为null
                        "OrderID188",// key:自定义Key,可以用于去重,可为null
                        ("Hello MetaQ").getBytes());// body:消息内容
// 发送消息并返回结果
SendResult sendResult = producer.send(msg);
// 清理资源,关闭网络连接,注销自己
producer.shutdown();

在整个应用生命周期内,生产者需要调用一次start方法来初始化,初始化主要完成的任务有:

  1. 如果没有指定namesrv地址,将会自动寻址
  2. 启动定时任务:更新namesrv地址、从namsrv更新topic路由信息、清理已经挂掉的broker、向所有broker发送心跳...
  3. 启动负载均衡的服务

初始化完成后,开始发送消息,发送消息的主要代码如下:

private SendResult sendDefaultImpl(Message msg,......) {
    // 检查Producer的状态是否是RUNNING
    this.makeSureStateOK();
    // 检查msg是否合法:是否为null、topic,body是否为空、body是否超长
    Validators.checkMessage(msg, this.defaultMQProducer);
    // 获取topic路由信息
    TopicPublishInfo topicPublishInfo = this.tryToFindTopicPublishInfo(msg.getTopic());
    // 从路由信息中选择一个消息队列
    MessageQueue mq = topicPublishInfo.selectOneMessageQueue(lastBrokerName);
    // 将消息发送到该队列上去
    sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, timeout);
}

代码中需要关注的两个方法tryToFindTopicPublishInfoselectOneMessageQueue。前面说过在producer初始化时,会启动定时任务获取路由信息并更新到本地缓存,所以tryToFindTopicPublishInfo会首先从缓存中获取topic路由信息,如果没有获取到,则会自己去namesrv获取路由信息。selectOneMessageQueue方法通过轮询的方式,返回一个队列,以达到负载均衡的目的。

如果Producer发送消息失败,会自动重试,重试的策略:

  1. 重试次数 < retryTimesWhenSendFailed(可配置)
  2. 总的耗时(包含重试n次的耗时) < sendMsgTimeout(发送消息时传入的参数)
  3. 同时满足上面两个条件后,Producer会选择另外一个队列发送消息
五、消息存储

RocketMQ的消息存储是由consume queuecommit log配合完成的。

1、Consume Queue

consume queue是消息的逻辑队列,相当于字典的目录,用来指定消息在物理文件commit log上的位置。

我们可以在配置中指定consumequeuecommitlog存储的目录
每个topic下的每个queue都有一个对应的consumequeue文件,比如:

${rocketmq.home}/store/consumequeue/${topicName}/${queueId}/${fileName}

Consume Queue文件组织,如图所示:

Consume Queue文件组织示意图
  1. 根据topicqueueId来组织文件,图中TopicA有两个队列0,1,那么TopicA和QueueId=0组成一个ConsumeQueue,TopicA和QueueId=1组成另一个ConsumeQueue。
  2. 按照消费端的GroupName来分组重试队列,如果消费端消费失败,消息将被发往重试队列中,比如图中的%RETRY%ConsumerGroupA
  3. 按照消费端的GroupName来分组死信队列,如果消费端消费失败,并重试指定次数后,仍然失败,则发往死信队列,比如图中的%DLQ%ConsumerGroupA

死信队列(Dead Letter Queue)一般用于存放由于某种原因无法传递的消息,比如处理失败或者已经过期的消息。

Consume Queue中存储单元是一个20字节定长的二进制数据,顺序写顺序读,如下图所示:

consumequeue文件存储单元格式
  1. CommitLog Offset是指这条消息在Commit Log文件中的实际偏移量
  2. Size存储中消息的大小
  3. Message Tag HashCode存储消息的Tag的哈希值:主要用于订阅时消息过滤(订阅时如果指定了Tag,会根据HashCode来快速查找到订阅的消息)
2、Commit Log

CommitLog:消息存放的物理文件,每台broker上的commitlog被本机所有的queue共享,不做任何区分。
文件的默认位置如下,仍然可通过配置文件修改:

${user.home} \store\${commitlog}\${fileName}

CommitLog的消息存储单元长度不固定,文件顺序写,随机读。消息的存储结构如下表所示,按照编号顺序以及编号对应的内容依次存储。

Commit Log存储单元结构图
3、消息存储实现

消息存储实现,比较复杂,也值得大家深入了解,后面会单独成文来分析(目前正在收集素材),这小节只以代码说明一下具体的流程。

// Set the storage time
msg.setStoreTimestamp(System.currentTimeMillis());
// Set the message body BODY CRC (consider the most appropriate setting
msg.setBodyCRC(UtilAll.crc32(msg.getBody()));
StoreStatsService storeStatsService = this.defaultMessageStore.getStoreStatsService();
synchronized (this) {
    long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
    // Here settings are stored timestamp, in order to ensure an orderly global
    msg.setStoreTimestamp(beginLockTimestamp);
    // MapedFile:操作物理文件在内存中的映射以及将内存数据持久化到物理文件中
    MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile();
    // 将Message追加到文件commitlog
    result = mapedFile.appendMessage(msg, this.appendMessageCallback);
    switch (result.getStatus()) {
    case PUT_OK:break;
    case END_OF_FILE:
         // Create a new file, re-write the message
         mapedFile = this.mapedFileQueue.getLastMapedFile();
         result = mapedFile.appendMessage(msg, this.appendMessageCallback);
     break;
     DispatchRequest dispatchRequest = new DispatchRequest(
                topic,// 1
                queueId,// 2
                result.getWroteOffset(),// 3
                result.getWroteBytes(),// 4
                tagsCode,// 5
                msg.getStoreTimestamp(),// 6
                result.getLogicsOffset(),// 7
                msg.getKeys(),// 8
                /**
                 * Transaction
                 */
                msg.getSysFlag(),// 9
                msg.getPreparedTransactionOffset());// 10
    // 1.分发消息位置到ConsumeQueue
    // 2.分发到IndexService建立索引
    this.defaultMessageStore.putDispatchRequest(dispatchRequest);
}
4、消息的索引文件

如果一个消息包含key值的话,会使用IndexFile存储消息索引,文件的内容结构如图:

消息索引

索引文件主要用于根据key来查询消息的,流程主要是:

  1. 根据查询的 key 的 hashcode%slotNum 得到具体的槽的位置(slotNum 是一个索引文件里面包含的最大槽的数目,例如图中所示 slotNum=5000000)
  2. 根据 slotValue(slot 位置对应的值)查找到索引项列表的最后一项(倒序排列,slotValue 总是指向最新的一个索引项)
  3. 遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的 32 条记录)
六、消息订阅

RocketMQ消息订阅有两种模式,一种是Push模式,即MQServer主动向消费端推送;另外一种是Pull模式,即消费端在需要时,主动到MQServer拉取。但在具体实现时,Push和Pull模式都是采用消费端主动拉取的方式。

首先看下消费端的负载均衡:

消费端负载均衡

消费端会通过RebalanceService线程,10秒钟做一次基于topic下的所有队列负载:

  1. 遍历Consumer下的所有topic,然后根据topic订阅所有的消息
  2. 获取同一topic和Consumer Group下的所有Consumer
  3. 然后根据具体的分配策略来分配消费队列,分配的策略包含:平均分配、消费端配置等

如同上图所示:如果有 5 个队列,2 个 consumer,那么第一个 Consumer 消费 3 个队列,第二 consumer 消费 2 个队列。这里采用的就是平均分配策略,它类似于分页的过程,TOPIC下面的所有queue就是记录,Consumer的个数就相当于总的页数,那么每页有多少条记录,就类似于某个Consumer会消费哪些队列。

通过这样的策略来达到大体上的平均消费,这样的设计也可以很方面的水平扩展Consumer来提高消费能力。

消费端的Push模式是通过长轮询的模式来实现的,就如同下图:

Push模式示意图

Consumer端每隔一段时间主动向broker发送拉消息请求,broker在收到Pull请求后,如果有消息就立即返回数据,Consumer端收到返回的消息后,再回调消费者设置的Listener方法。如果broker在收到Pull请求时,消息队列里没有数据,broker端会阻塞请求直到有数据传递或超时才返回。

当然,Consumer端是通过一个线程将阻塞队列LinkedBlockingQueue<PullRequest>中的PullRequest发送到broker拉取消息,以防止Consumer一致被阻塞。而Broker端,在接收到Consumer的PullRequest时,如果发现没有消息,就会把PullRequest扔到ConcurrentHashMap中缓存起来。broker在启动时,会启动一个线程不停的从ConcurrentHashMap取出PullRequest检查,直到有数据返回。

七、RocketMQ的其他特性

前面的6个特性都是基本上都是点到为止,想要深入了解,还需要大家多多查看源码,多多在实际中运用。当然除了已经提到的特性外,RocketMQ还支持:

  1. 定时消息
  2. 消息的刷盘策略
  3. 主动同步策略:同步双写、异步复制
  4. 海量消息堆积能力
  5. 高效通信
  6. .......

其中涉及到的很多设计思路和解决方法都值得我们深入研究:

  1. 消息的存储设计:既要满足海量消息的堆积能力,又要满足极快的查询效率,还要保证写入的效率。
  2. 高效的通信组件设计:高吞吐量,毫秒级的消息投递能力都离不开高效的通信。
  3. .......

RocketMQ最佳实践

一、Producer最佳实践

1、一个应用尽可能用一个 Topic,消息子类型用 tags 来标识,tags 可以由应用自由设置。只有发送消息设置了tags,消费方在订阅消息时,才可以利用 tags 在 broker 做消息过滤。
2、每个消息在业务层面的唯一标识码,要设置到 keys 字段,方便将来定位消息丢失问题。由于是哈希索引,请务必保证 key 尽可能唯一,这样可以避免潜在的哈希冲突。
3、消息发送成功或者失败,要打印消息日志,务必要打印 sendresult 和 key 字段。
4、对于消息不可丢失应用,务必要有消息重发机制。例如:消息发送失败,存储到数据库,能有定时程序尝试重发或者人工触发重发。
5、某些应用如果不关注消息是否发送成功,请直接使用sendOneWay方法发送消息。

二、Consumer最佳实践

1、消费过程要做到幂等(即消费端去重)
2、尽量使用批量方式消费方式,可以很大程度上提高消费吞吐量。
3、优化每条消息消费过程

三、其他配置

线上应该关闭autoCreateTopicEnable,即在配置文件中将其设置为false

RocketMQ在发送消息时,会首先获取路由信息。如果是新的消息,由于MQServer上面还没有创建对应的Topic,这个时候,如果上面的配置打开的话,会返回默认TOPIC的(RocketMQ会在每台broker上面创建名为TBW102的TOPIC)路由信息,然后Producer会选择一台Broker发送消息,选中的broker在存储消息时,发现消息的topic还没有创建,就会自动创建topic。后果就是:以后所有该TOPIC的消息,都将发送到这台broker上,达不到负载均衡的目的。

所以基于目前RocketMQ的设计,建议关闭自动创建TOPIC的功能,然后根据消息量的大小,手动创建TOPIC。

RocketMQ设计相关

RocketMQ的设计假定:

每台PC机器都可能宕机不可服务
任意集群都有可能处理能力不足
最坏的情况一定会发生
内网环境需要低延迟来提供最佳用户体验

RocketMQ的关键设计:

分布式集群化
强数据安全
海量数据堆积
毫秒级投递延迟(推拉模式)

这是RocketMQ在设计时的假定前提以及需要到达的效果。我想这些假定适用于所有的系统设计。随着我们系统的服务的增多,每位开发者都要注意自己的程序是否存在单点故障,如果挂了应该怎么恢复、能不能很好的水平扩展、对外的接口是否足够高效、自己管理的数据是否足够安全...... 多多规范自己的设计,才能开发出高效健壮的程序。

来源: http://www.jianshu.com/p/453c6e7ff81c

 

RocketMQ安装与使用

一、服务端安装部署

我是在虚拟机中的CentOS6.5中进行部署。
2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解压到适当的目录如/opt/目录
3.启动RocketMQ:进入rocketmq/bin 目录 执行
nohup sh mqnamesrv &

4.启动Broker,设置对应的NameServer

 nohup sh mqbroker -n "127.0.0.1:9876" &

二、编写客户端

可以查看sameple中的quickstart源码 1.Consumer 消息消费者

/**
* Consumer,订阅消息
*/
public class Consumer {

public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("QuickStartConsumer");

consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.setInstanceName("QuickStartConsumer");
consumer.subscribe("QuickStart", "*");

consumer.registerMessageListener(new MessageListenerConcurrently() {

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});

consumer.start();

System.out.println("Consumer Started.");
}
}
2.Producer消息生产者

/**
* Producer,发送消息
*
*/
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("QuickStartProducer");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setInstanceName("QuickStartProducer");
producer.start();

for (int i = 0; i < 1000; i++) {
try {
Message msg = new Message("QuickStart",// topic
"TagA",// tag
("Hello RocketMQ ,QuickStart" + i).getBytes()// body
);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}

producer.shutdown();
}
}

 

3.首先运行Consumer程序,一直在运行状态接收服务器端推送过来的消息

23:18:07.587 [main] DEBUG i.n.c.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16
23:18:07.591 [main] DEBUG i.n.util.internal.PlatformDependent - Platform: Windows
23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - Java version: 7
23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noUnsafe: false
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.ByteBuffer.cleaner: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Buffer.address: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: true
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - sun.misc.Unsafe: available
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noJavassist: false
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - Javassist: unavailable
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - You don't have Javassist in your class path or you don't have enough permission to load dynamically generated classes. Please check the configuration for better performance.
23:18:07.595 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false
23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false
23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512
23:18:08.355 [main] DEBUG i.n.util.internal.ThreadLocalRandom - -Dio.netty.initialSeedUniquifier: 0x8c0d4793e5820c31
23:18:08.446 [NettyClientWorkerThread_1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.noResourceLeakDetection: false
Consumer Started.

4.再次运行Producer程序,生成消息并发送到Broker,Producer的日志冲没了,但是可以看到Broker推送到Consumer的一条消息
1

ConsumeMessageThread-QuickStartConsumer-3 Receive New Messages: [MessageExt [queueId=0, storeSize=150, queueOffset=244, sysFlag=0, bornTimestamp=1400772029972, bornHost=/10.162.0.7:54234, storeTimestamp=1400772016017, storeHost=/127.0.0.1:10911, msgId=0A0A0A5900002A9F0000000000063257, commitLogOffset=406103, bodyCRC=112549959, reconsumeTimes=0, preparedTransactionOffset=0, toString()=Message [topic=QuickStart, flag=0, properties={TAGS=TagA, WAIT=true, MAX_OFFSET=245, MIN_OFFSET=0}, body=29]]]

三、Consumer最佳实践

1.消费过程要做到幂等(即消费端去重)

RocketMQ无法做到消息重复,所以如果业务对消息重复非常敏感,务必要在业务层面去重,有以下一些方式:

(1).将消息的唯一键,可以是MsgId,也可以是消息内容中的唯一标识字段,例如订单ID,消费之前判断是否在DB或Tair(全局KV存储) 中存在,如果不存在则插入,并消费,否则跳过。(实践过程要考虑原子性问题,判断是否存在可以尝试插入,如果报主键冲突,则插入失败,直接跳过) msgid一定是全局唯一的标识符,但是可能会存在同样的消息有两个不同的msgid的情况(有多种原因),这种情况可能会使业务上重复,建议最好使用消 息体中的唯一标识字段去重

(2).使业务层面的状态机去重

2.批量方式消费

如果业务流程支持批量方式消费,则可以很大程度上的提高吞吐量,可以通过设置Consumer的consumerMessageBatchMaxSize参数,默认是1,即一次消费一条参数

3.跳过非重要的消息

发生消息堆积时,如果消费速度一直跟不上发送速度,可以选择丢弃不重要的消息

@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);

long offset=msgs.get(0).getQueueOffset();

String maxOffset=msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
long diff=Long.parseLong(maxOffset)-offset;
if(diff>100000){
//处理消息堆积情况
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

如以上代码所示,当某个队列的消息数堆积到 100000 条以上,则尝试丢弃部分或全部消息,这样就可以快速追上发送消息的速度

4.优化没条消息消费过程

举例如下,某条消息的消费过程如下

1. 根据消息从 DB 查询数据 1

2. 根据消息从 DB 查询数据2

3. 复杂的业务计算

4. 向 DB 插入数据3

5. 向 DB 插入数据 4

这条消息的消费过程与 DB 交互了 4 次,如果按照每次 5ms 计算,那么总共耗时 20ms,假设业务计算耗时 5ms,那么总过耗时 25ms,如果能把 4 次 DB 交互优化为 2 次,那么总耗时就可以优化到 15ms,也就是说总体性能提高了 40%。

对于 Mysql 等 DB,如果部署在磁盘,那么与 DB 进行交互,如果数据没有命中 cache,每次交互的 RT 会直线上升, 如果采用 SSD,则 RT 上升趋势要明显好于磁盘。

个别应用可能会遇到这种情况:在线下压测消费过程中,db 表现非常好,每次 RT 都很短,但是上线运行一段时间,RT 就会变长,消费吞吐量直线下降

主要原因是线下压测时间过短,线上运行一段时间后,cache 命中率下降,那么 RT 就会增加。建议在线下压测时,要测试足够长时间,尽可能模拟线上环境,压测过程中,数据的分布也很重要,数据不同,可能 cache 的命中率也会完全不同

四、Producer最佳实践

1.发送消息注意事项

(1) 一个应用尽可能用一个 Topic,消息子类型用 tags 来标识,tags 可以由应用自由设置。只有发送消息设置了tags,消费方在订阅消息时,才可以利用 tags 在 broker 做消息过滤。

(2)每个消息在业务层面的唯一标识码,要设置到 keys 字段,方便将来定位消息丢失问题。服务器会为每个消息创建索引(哈希索引),应用可以 通过 topic,key 来查询这条消息内容,以及消息被谁消费。由于是哈希索引,请务必保证 key 尽可能唯一,这样可以避免潜在的哈希冲突。

(3)消息发送成功或者失败,要打印消息日志,务必要打印 sendresult 和 key 字段

(4)send 消息方法,只要不抛异常,就代表发送成功。但是发送成功会有多个状态,在 sendResult 里定义

  • SEND_OK:消息发送成功
  • FLUSH_DISK_TIMEOUT:消息发送成功,但是服务器刷盘超时,消息已经进入服务器队列,只有此时服务器宕机,消息才会丢失
  • ​FLUSH_SLAVE_TIMEOUT:消息发送成功,但是服务器同步到 Slave 时超时,消息已经进入服务器队列,只有此时服务器宕机,消息才会丢失
  • SLAVE_NOT_AVAILABLE:消息发送成功,但是此时 slave 不可用,消息已经进入服务器队列,只有此时服务器宕机,消息才会 丢失。对于精确发送顺序消息的应用,由于顺序消息的局限性,可能会涉及到主备自动切换问题,所以如果sendresult 中的 status 字段不等 于 SEND_OK,就应该尝试重试。对于其他应用,则没有必要这样

(5)对于消息不可丢失应用,务必要有消息重发机制

2.消息发送失败处理

Producer 的 send 方法本身支持内部重试,重试逻辑如下:

(1) 至多重试 3 次

(2) 如果发送失败,则轮转到下一个 Broker

(3) 这个方法的总耗时时间不超过 sendMsgTimeout 设置的值,默认 10s所以,如果本身向 broker 发送消息产生超时异常,就不会再做重试

如果调用 send 同步方法发送失败,则尝试将消息存储到 db,由后台线程定时重试,保证消息一定到达 Broker。

上述 db 重试方式为什么没有集成到 MQ 客户端内部做,而是要求应用自己去完成,基于以下几点考虑:

(1)MQ 的客户端设计为无状态模式,方便任意的水平扩展,且对机器资源的消耗仅仅是 cpu、内存、网络

(2)如果 MQ 客户端内部集成一个 KV 存储模块,那么数据只有同步落盘才能较可靠,而同步落盘本身性能开销较大,所以通常会采用异步落盘, 又由于应用关闭过程不受 MQ 运维人员控制,可能经常会发生 kill -9 这样暴力方式关闭,造成数据没有及时落盘而丢失

(3)Producer 所在机器的可靠性较低,一般为虚拟机,不适合存储重要数据。 综上,建议重试过程交由应用来控制。

3.选择 oneway 形式发送

一个 RPC 调用,通常是这样一个过程

(1)客户端发送请求到服务器

(2)服务器处理该请求

(3)服务器向客户端返回应答

所以一个 RPC 的耗时时间是上述三个步骤的总和,而某些场景要求耗时非常短,但是对可靠性要求并不高,例如日志收集类应用,此类应用可以采 用 oneway 形式调用,oneway 形式只发送请求不等待应答,而发送请求在客户端实现层面仅仅是一个 os 系统调用的开销,即将数据写入客户 端的 socket 缓冲区,此过程耗时通常在微秒级

 

RocketMQ不止可以直接推送消息,在消费端注册监听器进行监听,还可以由消费端决定自己去拉取数据
/**
* PullConsumer,订阅消息
*/
public class PullConsumer {
//Java缓存
private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();

public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PullConsumerGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.start();
//拉取订阅主题的队列,默认队列大小是4
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTestMapBody");
for (MessageQueue mq : mqs) {
System.out.println("Consume from the queue: " + mq);
SINGLE_MQ:while(true){
try {

PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
List<MessageExt> list=pullResult.getMsgFoundList();
if(list!=null&&list.size()<100){
for(MessageExt msg:list){
System.out.println(SerializableInterface.deserialize(msg.getBody()));
}
}
System.out.println(pullResult.getNextBeginOffset());
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());

switch (pullResult.getPullStatus()) {
case FOUND:
// TODO
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
}
catch (Exception e) {
e.printStackTrace();
}
}
}

consumer.shutdown();
}

private static void putMessageQueueOffset(MessageQueue mq, long offset) {
offseTable.put(mq, offset);
}

private static long getMessageQueueOffset(MessageQueue mq) {
Long offset = offseTable.get(mq);
if (offset != null){
System.out.println(offset);
return offset;
}
return 0;
}

/**
* PullConsumer,订阅消息
*/
public class PullConsumer {
//Java缓存
private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();

public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PullConsumerGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.start();
//拉取订阅主题的队列,默认队列大小是4
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTestMapBody");
for (MessageQueue mq : mqs) {
System.out.println("Consume from the queue: " + mq);
SINGLE_MQ:while(true){
try {

PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
List<MessageExt> list=pullResult.getMsgFoundList();
if(list!=null&&list.size()<100){
for(MessageExt msg:list){
System.out.println(SerializableInterface.deserialize(msg.getBody()));
}
}
System.out.println(pullResult.getNextBeginOffset());
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());

switch (pullResult.getPullStatus()) {
case FOUND:
// TODO
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
}
catch (Exception e) {
e.printStackTrace();
}
}
}

consumer.shutdown();
}

private static void putMessageQueueOffset(MessageQueue mq, long offset) {
offseTable.put(mq, offset);
}

private static long getMessageQueueOffset(MessageQueue mq) {
Long offset = offseTable.get(mq);
if (offset != null){
System.out.println(offset);
return offset;
}
return 0;
}

刚开始的没有细看PullResult对象,以为拉取到的结果没有MessageExt对象还跑到群里面问别人,犯2了

特别要注意  静态变量offsetTable的作用,拉取的是按照从offset(理解为下标)位置开始拉取,拉取N条,offsetTable记录下次拉取的offset位置

 

来源:http://www.changeself.net/archives/rocketmq%E5%85%A5%E9%97%A8%EF%BC%882%EF%BC%89%E6%9C%80%E4%BD%B3%E5%AE%9E%E8%B7%B5.html

RocketMQ入门

RocketMQ是一款分布式、队列模型的消息中间件,具有以下特点:能够保证严格的消息顺序

一.RocketMQ网络部署特点

来源:http://www.changeself.net/archives/rocketmq%E5%85%A5%E9%97%A8%EF%BC%881%EF%BC%89.html

    (1)NameServer是一个几乎无状态的节点,可集群部署,节点之间无任何信息同步
    (2)Broker部署相对复杂,Broker氛围Master与Slave,一个Master可以对应多个Slaver,但是一个Slaver只能对应 一个Master,Master与Slaver的对应关系通过指定相同的BrokerName,不同的BrokerId来定义,BrokerId为0表示 Master,非0表示Slaver。Master可以部署多个。每个Broker与NameServer集群中的所有节点建立长连接,定时注册 Topic信息到所有的NameServer
    (3)Producer与NameServer集群中的其中一个节点(随机选择)建立长连接,定期从NameServer取Topic路由信息,并向提供 Topic服务的Master建立长连接,且定时向Master发送心跳。Produce完全无状态,可集群部署
    (4)Consumer与NameServer集群中的其中一个节点(随机选择)建立长连接,定期从NameServer取Topic路由信息,并向提供 Topic服务的Master、Slaver建立长连接,且定时向Master、Slaver发送心跳。Consumer即可从Master订阅消息,也 可以从Slave订阅消息,订阅规则由Broker配置决定

二.RocketMQ储存特点

   (1)零拷贝原理:Consumer消费消息过程,使用了零拷贝,零拷贝包括一下2中方式,RocketMQ使用第一种方式,因小块数据传输的要求效果比sendfile方式好
           a )使用mmap+write方式
             优点:即使频繁调用,使用小文件块传输,效率也很高
             缺点:不能很好的利用DMA方式,会比sendfile多消耗CPU资源,内存安全性控制复杂,需要避免JVM Crash问题
        b)使用sendfile方式
             优点:可以利用DMA方式,消耗CPU资源少,大块文件传输效率高,无内存安全新问题
             缺点:小块文件效率低于mmap方式,只能是BIO方式传输,不能使用NIO
    (2)数据存储结构

三.RocketMQ关键特性

1.单机支持1W以上的持久化队列

    (1)所有数据单独储存到commit Log ,完全顺序写,随机读
    (2)对最终用户展现的队列实际只储存消息在Commit Log 的位置信息,并且串行方式刷盘
     这样做的好处:
    (1)队列轻量化,单个队列数据量非常少
    (2)对磁盘的访问串行话,避免磁盘竞争,不会因为队列增加导致IOWait增高
     每个方案都有优缺点,他的缺点是:
    (1)写虽然是顺序写,但是读却变成了随机读
    (2)读一条消息,会先读Consume  Queue,再读Commit Log,增加了开销
    (3)要保证Commit Log 与 Consume  Queue完全的一致,增加了编程的复杂度
     以上缺点如何客服:
    (1)随机读,尽可能让读命中pagecache,减少IO操作,所以内存越大越好。如果系统中堆积的消息过多,读数据要访问硬盘会不会由于随机读导致系统性能急剧下降,答案是否定的。
        a)访问pagecache时,即使只访问1K的消息,系统也会提前预读出更多的数据,在下次读时就可能命中pagecache
        b)随机访问Commit Log 磁盘数据,系统IO调度算法设置为NOOP方式,会在一定程度上将完全的随机读变成顺序跳跃方式,而顺序跳跃方式读较完全的随机读性能高5倍
    (2)由于Consume Queue存储数量极少,而且顺序读,在pagecache的与读取情况下,Consume Queue的读性能与内存几乎一直,即使堆积情况下。所以可以认为Consume Queue完全不会阻碍读性能
    (3)Commit Log中存储了所有的元信息,包含消息体,类似于MySQl、Oracle的redolog,所以只要有Commit Log存在, Consume  Queue即使丢失数据,仍可以恢复出来

2.刷盘策略

rocketmq中的所有消息都是持久化的,先写入系统pagecache,然后刷盘,可以保证内存与磁盘都有一份数据,访问时,可以直接从内存读取
2.1异步刷盘
在有 RAID 卡, SAS 15000 转磁盘测试顺序写文件,速度可以达到 300M 每秒左右,而线上的网卡一般都为千兆网卡,写磁盘速度明显快于数据网络入口速度,那么是否可以做到写完    内存就向用户返回,由后台线程刷盘呢?

(1).  由于磁盘速度大于网卡速度,那么刷盘的进度肯定可以跟上消息的写入速度。

(2).  万一由于此时系统压力过大,可能堆积消息,除了写入 IO,还有读取 IO,万一出现磁盘读取落后情况,会不会导致系统内存溢出,答案是否定的,原因如下:

a)  写入消息到 PAGECACHE 时,如果内存不足,则尝试丢弃干净的 PAGE,腾出内存供新消息使用,策略

是 LRU 方式。

b)  如果干净页不足,此时写入 PAGECACHE 会被阻塞,系统尝试刷盘部分数据,大约每次尝试 32 个 PAGE,

来找出更多干净 PAGE。

综上,内存溢出的情况不会出现

2.2同步刷盘:
同步刷盘与异步刷盘的唯一区别是异步刷盘写完 PAGECACHE 直接返回,而同步刷盘需要等待刷盘完成才返回,同步刷盘流程如下:
    (1)写入 PAGECACHE 后,线程等待,通知刷盘线程刷盘。
    (2)刷盘线程刷盘后,唤醒前端等待线程,可能是一批线程。
    (3)前端等待线程向用户返回成功。

3.消息查询

3.1按照MessageId查询消息

MsgId总共16个字节,包含消息储存主机地址,消息Commit Log Offset。从MsgId中解析出Broker的地址和Commit Log 偏移地址,然后按照存储格式所在位置消息buffer解析成一个完整消息
3.2按照Message Key查询消息

1.根据查询的key的hashcode%slotNum得到具体的槽位置  (slotNum是一个索引文件里面包含的最大槽目数目,例如图中所示 slotNum=500W)
2.根据slotValue(slot对应位置的值)查找到索引项列表的最后一项(倒序排列,slotValue总是指向最新的一个索引项)
3.遍历索引项列表返回查询时间范围内的结果集(默认一次最大返回的32条记录)
4.Hash冲突,寻找key的slot位置时相当于执行了两次散列函数,一次key的hash,一次key的hash取值模,因此这里存在两 次冲突的情况;第一种,key的hash值不同但模数相同,此时查询的时候会在比较第一次key的hash值(每个索引项保存了key的hash值),过 滤掉hash值不想等的情况。第二种,hash值相等key不想等,出于性能的考虑冲突的检测放到客户端处理(key的原始值是存储在消息文件中的,避免 对数据文件的解析),客户端比较一次消息体的key是否相同
5.存储,为了节省空间索引项中存储的时间是时间差值(存储时间——开始时间,开始时间存储在索引文件头中),整个索引文件是定长的,结构也是固定的

4.服务器消息过滤

RocketMQ的消息过滤方式有别于其他的消息中间件,是在订阅时,再做过滤,先来看下Consume Queue存储结构

1.在Broker端进行Message Tag比较,先遍历Consume Queue,如果存储的Message Tag与订阅的Message Tag不符合,则跳过,继续比对下一个,符合则传输给Consumer。注意Message Tag是字符串形式,Consume Queue中存储的是其对应的hashcode,比对时也是比对hashcode
2.Consumer收到过滤消息后,同样也要执行在broker端的操作,但是比对的是真实的Message Tag字符串,而不是hashcode
为什么过滤要这么做?
1.Message Tag存储hashcode,是为了在Consume Queue定长方式存储,节约空间
2.过滤过程中不会访问Commit Log 数据,可以保证堆积情况下也能高效过滤
3.即使存在hash冲突,也可以在Consumer端进行修正,保证万无一失

5.单个JVM进程也能利用机器超大内存

1.Producer发送消息,消息从socket进入java 堆
2.Producer发送消息,消息从java堆进入pagecache,物理内存
3.Producer发送消息,由异步线程刷盘,消息从pagecache刷入磁盘
4.Consumer拉消息(正常消费),消息直接从pagecache(数据在物理内存)转入socket,到达Consumer,不经过java堆。这种消费场景最多,线上96G物理内存,按照1K消息算,可以物理缓存1亿条消息
5.Consumer拉消息(异常消费),消息直接从pagecache转入socket
6.Consumer拉消息(异常消费),由于socket访问了虚拟内存,产生缺页中断,此时会产生磁盘IO,从磁盘Load消息到pagecache,然后直接从socket发出去
7.同5
8.同6

6.消息堆积问题解决办法

堆积性能指标
1消息的堆积容量依赖磁盘大小
2发消息的吞吐量大小受影响程度无Slave情况,会受一定影响
有Slave情况,不受影响
3正常消费的Consumer是否会受影响无Slave情况,会受一定影响
有Slave情况,不受影响
4访问堆积在磁盘的消息时,吞吐量有多大与访问的并发有关,最终会降到5000左右
在有Slave情况下,Master一旦发现Consumer访问堆积在磁盘的数据时,回想Consumer下达一个重定向指令,令 Consumer从Slave拉取数据,这样正常的发消息与正常的消费不会因为堆积受影响,因为系统将堆积场景与非堆积场景分割在了两个不同的节点处理。 这里会产生一个问题,Slave会不会写性能下降,答案是否定的。因为Slave的消息写入只追求吞吐量,不追求实时性,只要整体的吞吐量高就行了,而 Slave每次都是从Master拉取一批数据,如1M,这种批量顺序写入方式使堆积情况,整体吞吐量影响相对较小,只是写入RT会变长