Kafka&MetaQ

花花世界蕾

Memorphosis介绍：
　　Memorphosis是一个消息中间件，它是linkedin开源MQ------kafka的Java版本，针对淘宝内部应用做了定制和优化。

Memorphosis做到了啥：
　　顺序（没做到，只能保证先进先出）
安全（部分如磁盘掉电会最多丢失1000条消息或10秒内信息）
性能（性能很好单台meta，10个同组256大小每秒4.5W）
消耗（消耗低单台meta，10个同组256大小每秒4.5W cpu 13% IO 0.15）
Metamorphosis的部署结构：


Metamorphosis的性能：




Metamorphosis的特点：
　　除了完整实现kafka的功能之外，我们还为meta加入了额外的功能，使得meta成为一个更为强大的通用消息中间件，包括

• 彻底用java重写的实现，高效的协议和通讯框架
  
• 发送端的负载均衡
  
• Master/Slave异步和同步复制的高可用方案
  
• 专门用于广播消息的客户端实现
  
• 与diamond结合使用的顺序发送消息功能
  
• 消息都是持久的，保存在磁盘(顺序写)吞吐量第一（1秒钟发送条数4k消息，2w）消费状态保存在客户端分布式，生产者、服务器和消费者都可分布
  

　　消息的实时性：
broker上配置的批量force消息的阈值，默认是1000条force一次。这个值越大，则实时性越低。
消费者每次抓取的数据大小，这个值越大，则实时性越低，但是吞吐量越高。
Topic的分区数目对实时性也有较大影响，分区数目越多，则磁盘压力越大，导致消息投递的实时性降低。
消费者重试抓取的时间间隔，越长则延迟越严重。
消费者抓取数据的线程数

　　保证消息不丢：
　　Meta的所有消息都将写入磁盘，没有区分所谓持久消息和非持久消息，所有消息都是持久的
生产者在发送消息的时候，服务端只在将消息写入磁盘的情况下才会返回应答给生产者告知消息写入成功
消费者在消费消息后，消息也不会删除，而是按照设定的过期处理策略进行批量处理，与任何消费者无关。消费者可在任何时候进行消费，也就是没有区分持久订阅和非持久订阅，都是持久订阅。
写入磁盘的消息，其实是写入os缓冲区，可根据你对数据可靠性的要求设定sync策略。
单机broker可能因为磁盘损坏等硬件故障永久丢失消息，meta还提供了高可用的HA复制方案，通过将消息复制到多个broker来提高数据可靠性。
注：
消息生产者发送的消息，meta服务器收到后在做必要的校验和检查之后的第一件事就是写入磁盘，写入成功之后返回应答给生产者。因此，可以确认每条发送结果为成功的消息服务器都是写入磁盘的。
写入磁盘，不意味着数据落到磁盘设备上，毕竟我们还隔着一层os，os对写有缓冲。Meta有两个特性来保证数据落到磁盘上
每1000条（可配置），即强制调用一次force来写入磁盘设备。
每隔10秒（可配置），强制调用一次force来写入磁盘设备。
因此,Meta通过配置可保证在异常情况下（如磁盘掉电）10秒内最多丢失1000条消息。当然通过参数调整你甚至可以在掉电情况下不丢失任何消息。

　　顺序：
Meta是磁盘的顺序存储，而消费的顺序是按照数据在磁盘文件上的前后输出的。
消息重复
1：针对生产者来说，有可能发生这种情况，生产者发送消息，等待服务器应答，这个时候发生网络故障，服务器实际已经将消息写入成功，
但是由于网络故障没有返回应答。那么生产者会认为发送失败，则再次发送同一条消息，如果发送成功，则服务器实际存储两条相同的消息。这种由故障引起的重复，
meta是无法避免的，因为meta不判断消息的data是否一致，因为它并不理解data的语义，而仅仅是作为载荷来传输。
2：针对消费者来说也有这个问题，消费者成功消费一条消息，但是此时断电，没有及时将前进后的offset存储起来，则下次启动的时候或者其他同个分组的消费者owner到这个分区的时候，
会重复消费该条消息。这种情况meta也无法完全避免。
　　

Metamorphosis现状：
　　服务器版本最高1.4.3
　　客户端 版本最高1.4.4
Meta的概念和术语介绍：
　　消息生产者
也称为Message Producer，一般简称为producer，负责产生消息并发送消息到meta服务器。
　　消息消费者
也称为Message Consumer，一般简称为consumer，负责消息的消费，meta采用pull模型，由消费者主动从meta服务器拉取数据并解析成消息并消费。
　　Topic
消息的主题，由用户定义并在服务端配置。producer发送消息到某个topic下，consumer从某个topic下消费消息。
　　分区(partition)
同一个topic下面还分为多个分区，如meta-test这个topic我们可以分为10个分区，分别有两台服务器提供，那么可能每台服务器提供5个分区，假设服务器id分别为0和1，则所有分区为0-0、0-1、0-2、0-3、0-4、1-0、1-1、1-2、1-3、1-4。
　　分区跟消费者的负载均衡机制有很大关系，具体见集群和负载均衡。
　　Message
消息，负载用户数据并在生产者、服务端和消费者之间传输。
　　Broker
就是meta的服务端或者说服务器，在消息中间件中也通常称为broker。
　　消费者分组(Group)
消费者可以是多个消费者共同消费一个topic下的消息，每个消费者消费部分消息。这些消费者就组成一个分组，拥有同一个分组名称,通常也称为消费者集群
　　Offset
消息在broker上的每个分区都是组织成一个文件列表，消费者拉取数据需要知道数据在文件中的偏移量，这个偏移量就是所谓offset。Offset是绝对偏移量，服务器会将offset转化为具体文件的相对偏移量。详细内容参见#消息的存储结构
负载均衡：
　　负载均衡和failover分不开，我们将分别讨论下生产者和消费者的负载均衡策略。我们先假定broker是一个集群，这样每个topic必定有多个分区。
　　生产者的负载均衡和failover ：
　　每个broker都可以配置一个topic可以有多少个分区，但是在生产者看来，一个topic在所有broker上的的所有分区组成一个分区列表来使用。
　　在创建producer的时候，客户端会从zookeeper上获取publish的topic对应的broker和分区列表，生产者在发送消息的时候必须选择一台broker上的一个分区来发送消息，默认的策略是一个轮询的路由规则。

　　

　　生产者在通过zk获取分区列表之后，会按照brokerId和partition的顺序排列组织成一个有序的分区列表，发送的时候按照从头到尾循环往复的方式选择一个分区来发送消息。考虑到我们的broker服务器软硬件配置基本一致，默认的轮询策略已然足够。
　　如果你想实现自己的负载均衡策略，可以实现上文提到过的PartitionSelector接口，并在创建producer的时候传入即可。
　　代码：
　　分区选择器：

import com.taobao.metamorphosis.client.producer; public class MyPartitionSelector implementsPartitionSelector { private final PositiveAtomicCounter sets = newPositiveAtomicCounter(); //partitions可用的分区的列表 //message消息的内容 @Override publicPartition getPartition(finalString topic, finalList<Partition> partitions, finalMessage message) throwsMetaClientException { if(partitions == null) { thrownew MetaClientException("There is no aviable partition for topic "+ topic + ",maybe you don't publish it at first?"); } try{ //或更具message内的字段进行分区的选择 如：id%partitions.size(); returnpartitions.get(this.sets.incrementAndGet() % partitions.size()); } catch (final Throwable t) { throw new MetaClientException(t); } } }

　　用选择器产生生产者：

MessageSessionFactory sessionFactory =new MetaMessageSessionFactory(metaClientConfig); sessionFactory.createProducer(newMyPartitionSelector());

　　

　　服务器配置：
　　conf/server.ini
　　brokerId=0
　　numPartitions=1(分区数)
　　分区的选择：
　　broker1：
　　brokerId=0
　　numPartitions=3(分区数)
　　broker2：
　　brokerId=1
　　numPartitions=3(分区数)
　　分区分别为0-1、0-2、 0-3、1-1、1-2、 1-3
　　消费者的负载均衡：
消费者的负载均衡会相对复杂一些。我们这里讨论的是单个分组内的消费者集群的负载均衡，不同分组的负载均衡互不干扰，没有讨论的必要。消费者的负载均衡跟topic的分区数目紧密相关，要考察几个场景。 首先是，单个分组内的消费者数目如果比总的分区数目多的话，则多出来的消费者不参与消费。

　　综上所述，单个分组内的消费者集群的负载均衡策略如下：
　　每个分区针对同一个group只挂载一个消费者
如果同一个group的消费者数目大于分区数目，则多出来的消费者将不参与消费
如果同一个group的消费者数目小于分区数目，则有部分消费者需要额外承担消费任务
Meta的客户端会自动帮处理消费者的负载均衡，它会将消费者列表和分区列表分别排序，然后按照上述规则做合理的挂载

Meta支持事务：
　　Metamorphosis 1.2开始支持事务，包括发送端和消费端事务。发送端同时支持本地事务和分布式事务，可以在一个事务内发送多条消息，要么同时成功，要么同时失败；可以使用XA事务，在事务内操作其他XA资源，例如操作数据库，与此同时发送meta消息，可以保证这些操作和发送消息要么一起成功，要么一起失败。
在消费消息的时候，可以批量消费一批消息，要么一起消费成功，要么失败重试。
　　代码：

try{ // 开始事务 producer.beginTransaction(); // 在事务内发送两条消息 if (!producer.sendMessage(newMessage(topic, line.getBytes())).isSuccess()) { // 发送失败，立即回滚 producer.rollback(); continue; } if (!producer.sendMessage(newMessage(topic, line.getBytes())).isSuccess()) { producer.rollback(); continue; } // 提交 producer.commit(); } catch (finalException e) { producer.rollback(); }

　　

服务端配置（server.ini）：
　　Meta服务端配置主要在服务器conf目录下的server.ini文件，整体配置分为三部分：系统参数、zookeeper参数以及topic配置。系统参数在system section，zookeeper参数配置在zookeeper section，而topic的配置是在topic=xxxx section。具体说明如下：
　　一份默认提供的参数配置在这里。
　　系统参数部分
系统参数配置都放在[system]下面：
　　brokerId: 服务器集群中唯一的id，必须为整型0-1024之间。对服务器集群的定义是使用同一个zookeeper并且在zookeeper上的root path相同，具体参见zookeeper配置。
hostName: 服务器hostname，默认取本机IP地址，如果你是多网卡机器，可能需要明确指定。服务器会将此hostname加上端口写入到zookeeper提供给客户端发现。
　　serverPort:服务器端口，默认8123。PS. 选择8123是因为这蕴含着我儿子的生日 :D。
　　numPartitions:系统默认情况下每个topic的分区数目，默认为1，可被topic配置覆盖。单个服务器的总分区数目不建议超过1000，太多将导致频繁的磁盘寻道严重影响IO性能。
　　dataPath: 服务器数据文件路径，默认在~home/meta下,每个topic可以覆盖此配置，对于多块磁盘的机器，可设置不同topic到不同磁盘来提升IO效率。
　　dataLogPath：数据日志文件路径，主要存放事务日志，默认跟dataPath一致，最好单独设置到不同的磁盘或者目录上。如果为空，使用指定的dataPath
　　getProcessThreadCount: 处理get请求的并发线程数，默认为CPUS*10。
　　putProcessThreadCount: 处理put请求的并发线程数，默认为CPUS*10。
　　maxSegmentSize: 单个数据文件的大小，默认为1G。默认无需修改此选项。
　　maxTransferSize: 传输给消费者的最大数据大小，默认为1M，请根据你的最大消息大小酌情设置，如果太小，每次无法传输一个完整的消息给消费者，导致消费者消费停滞。可设置成一个大数来取消限制。
　　1.4.3引入的新参数：
　　acceptPublish: 是否接收消息，默认为true；如果为false，则不会注册发送信息到zookeeper上，客户端当然无法发送消息到该broker。本参数可以被后续的topic配置覆盖。
　　acceptSubscribe: 与acceptPublish类似，默认也为true；如果为false，则不会注册消费信息到zookeeper上，消费者无法发现该broker，当然无法从该broker消费消息。本参数可以被后续的topic配置覆盖。
　　数据可靠性参数
Meta保证消息可靠性是建立在磁盘可靠性的基础上，发送的每一条消息都保证是在“写入磁盘”的情况下才返回给客户端应答。这里有两个关键参数可以控制：
　　unflushThreshold: 每隔多少条消息做一次磁盘sync，强制将更改的数据刷入磁盘。默认为1000。也就是说在掉电情况下，最多允许丢失1000条消息。可设置为0，强制每次写入都sync。在设置为0的情况下，服务器会自动启用group commit技术，将多个消息合并成一次sync来提升IO性能。经过测试，group commit情况下消息发送者的TPS没有受到太大影响，但是服务端的负载会上升很多。
unflushInterval: 间隔多少毫秒定期做一次磁盘sync,默认是10秒。也就是说在服务器掉电情况下，最多丢失10秒内发送过来的消息。不可设置为小于或者等于0。
请注意，上述两个参数都可以被topic单独配置说覆盖，也就是说每个topic可以配置不同的数据可靠级别。
　　数据删除策略配置
默认情况下，meta是会保存不断添加的消息，然后定期对“过期”的数据进行删除或者归档处理，这都是通过下列参数控制的：
　　deleteWhen: 何时执行删除策略的cron表达式，默认是0 0 6,18 * * ?，也就是每天的早晚6点执行处理策略。
deletePolicy: 数据删除策略，默认超过7天即删除,这里的168是小时，10s表示10秒，10m表示10分钟，10h表示10小时，不明确指定单位默认为小时。delete是指删除，超过指定时间的数据文件将被彻底从磁盘删除。也可以选择archive策略，即不对过期的数据文件做删除而是归档，当使用archive策略的时候可以选择是否压缩数据文件，如167,archive,true即选择将更改时间超过7天的数据文件归档并压缩为zip文件，如果不选择压缩，则重命名为扩展名为arc的文件。
上述两个参数都可以被topic单独配置所覆盖，也就是每个topic可以指定自己独特的删除策略。通常来说，对于不重要的topic可以将更早地将他们删除来节省磁盘空间。
　　事务相关配置
maxCheckpoints: 最大保存事务checkpoint数目，默认为3，服务器在启动的时候会从最近一次checkpoint回访事务日志文件，恢复重启前的事务状态。不建议修改此参数。
checkpointInterval：事务checkpoint时间间隔，单位毫秒，默认1小时。间隔时间太长，会导致启动的时候replay事务日志占用了太多时间，太短则可能影响到性能。
maxTxTimeoutTimerCapacity：最大事务超时timer的数量。服务端会为每个事务启动一个定时器监控事务是否超时，定时器的数目上限通过本参数限制。限制了本参数，也变相地控制了最大可运行的事务数。默认为30000个。
maxTxTimeoutInSeconds：最大事务超时时间，单位为秒，默认为60秒。客户端设置的事务超时时间不能超过此设定，超过将被强制限制为此设定。
flushTxLogAtCommit：服务端对事务日志的sync策略，0表示让操作系统决定，1表示每次commit都刷盘，2表示每隔1秒刷盘一次。此参数严重影响事务性能，可根据你需要的性能和可靠性之间权衡做出一个合理的选择。通常建议设置为2，表示每隔1秒刷盘一次，也就是最多丢失一秒内的运行时事务。这样的可靠级别对大多数服务是足够的。最安全的当然是设置为1，但是将严重影响事务性能。而0的安全级别最低。安全级别上 1>=2>0，而性能则是0 >= 2 > 1。
zookeeper配置
meta服务端会将自身id,topic信息和socket地址发送到zookeeper上，让客户端可以发现并连接服务器。Zookeeper相关的配置放在[zookeeper]模块下面:
　　zk.zkEnable: 是否启用zookeeper，也就是是否将信息注册到zookeeper上。默认为true。对于同步复制的slave来说，本参数会被强制设置为false。
zk.zkConnect: zookeeper服务器列表，例如localhost:1281这样的字符串。默认也是localhost:2181。请设置你的zk集群地址列表。
zk.zkSessionTimeoutMs: zookeeper的session timeout，默认为30秒。单位毫秒。
zk.zkConnectionTimeoutMs： zookeeper的连接超时时间，默认同样为30秒，单位毫秒。
zk.zkSyncTimeMs: 预期的zk集群间数据同步延迟，默认为5秒，这个参数对服务器无意义。
Topic配置
服务器将提供哪些topic服务都是通过topic配置来实现的，topic配置都是在[topic=xxx]的模块下面，其中xxx就是topic名称，一个示范配置如下：
　　[topic=boyan-test]
stat=true
numPartitions=1
这里配置了一个名为test的topic，并针对该topic启用实时统计，并将topic的在本服务器的分区数目设置为1。可见，topic配置可覆盖服务器的部分配置，包括：
　　stat:是否启用实时统计，启用则会在服务端对该topic的请求做实时统计，可以通过stats topic-name协议观察到该topic运行状况，可选。
numPartitions: 该topic在本服务器的分区总数，覆盖系统配置，可选。
unflushInterval：每隔多少条消息做一次磁盘sync，覆盖系统配置，可选。
unflushThreshold：每隔多少秒做一次磁盘sync，覆盖系统配置，可选。
deletePolicy：topic的删除策略，覆盖系统配置，可选。
deleteWhen：删除策略的执行时间，覆盖系统配置，可选。
dataPath：设置数据文件路径，覆盖系统配置，可选。
1.4.3新增参数：
　　acceptPublish: 是否接收该topic的消息，覆盖系统配置，可选。
acceptSubscribe: 是否接受消费者的订阅，覆盖系统配置，可选。
新增Topic热部署
在新增或者删除topic并保存server.ini之后，可以通过下列命令热加载新的配置文件并生效：
　　bin/metaServer.sh reload

服务器(1.4.3)部署：
　　master启动
　　配置一下：
　　brokerId=0（必须唯一）
　　zk.zkConnect=10.132.97.66:2181,10.132.97.73:2181
　　例子：

[system] brokerId=0 numPartitions=1 serverPort=8123 unflushThreshold=0 unflushInterval=10000 maxSegmentSize=1073741824 maxTransferSize=1048576 deletePolicy=delete,168 deleteWhen=00 6,18* * ? flushTxLogAtCommit=1 [zookeeper] zk.zkConnect=10.132.97.66:2181,10.132.97.73:2181 zk.zkSessionTimeoutMs=30000 zk.zkConnectionTimeoutMs=30000 zk.zkSyncTimeMs=5000 deletePolicy=delete,10h ;; Topics section [topic=test] stat=true [topic=meta-test] stat=true

　　启动服务：





　　./metaServer.sh start
　　

　　查看服务是否起来：

./metaServer.sh stats

　　关闭服务器：
?

./metaServer.sh stop

　　等它关闭后用 ps aux|grep meta查看是否完全关闭 ，不让在下次启动时会重复打开进程的。
　　启动多台master(修改serverPort端口和brokerId在server.ini)
　　启动（异步）slave：
　　server.ini 与master一致(在一台机器上的slave要修改jmx端口在metaServer.sh内，修改serverPort端口在server.ini)
　　配置async_slave.properties
　　slaveId=1（slaveId在一个master下必须一致）
　　例子：

#slave编号,大于等于0表示作为slave启动,同一个master下的slave编号应该设不同值. slaveId=1 #作为slave启动时向master订阅消息的group,如果没配置则默认为meta-slave-group slaveGroup=meta-slave-group #slave数据同步的最大延时,单位毫秒 slaveMaxDelayInMills=500 #是否自动从master同步server.ini #第一次仍然需要自己拷贝server.ini，后续可以通过设置此选项为true来自动同步 autoSyncMasterConfig=true

　　启动服务：

 ./metaServer.sh start slave 

　　启动后查看服务：
　　slave true
　　启动（同步）slave（1.4.4 ）：

publicclass GregorSlaveBrokerextends AbstractBrokerPlugin { OrderedThreadPoolExecutor orderedPutExecutor; @Override publicvoid start() { }

　　还没实现。

http方式访问：
　　Meta是通讯是走TCP长连接，它的的协议是基于文本行的协议，类似memcached的文本协议。通用的协议格式如下
　　开启访问 telnet localhost 8123
　　命令介绍：
put

put topic partition value-length flag [transactionKey]\r\n[body]

　　发送消息协议,topic为发送的消息主题，partition为发送的目的分区，value-length为发送的消息体长度,flag为消息标识位,transactionKey为事务标识符，可选。
　　示范：

put meta-test 0 5 0 1\r\nhello

　　

get

get topic group partition offset maxSize\r\n

　　消费者拉取消息协议，topic为拉取的消息主题，group为消费者分组名称，partition为拉取的目的分区，offset为拉取的起始偏移量，maxSize为本次拉取的最大数据量大小。
　　示范：

get meta-test example0 1024 512 1\r\n

　　

offset

offset topic group partition offset\r\n

　　查询离某个offset的最近有效的offset,topic为查询的消息主题，group为消费者分组名称，partition为查询的分区,offset为查询的offset，
　　示范：

offset meta-test example0 1024 1\r\n

stats

stats [item]\r\n

　　查询服务器的统计情况，item为查询的项目名称，如realtime(实时统计),具体的某个topic等，可以为空。
　　示范：

stats test\r\n

消息的存储结构：
　　消息在服务器的是按照顺序连续append在一起的，具体的单个消息的存储结构如下：
　　message length(4 bytes),包括消息属性和payload data
checksum(4 bytes)
message id(8 bytes)
message flag(4 bytes)
attribute length(4 bytes) + attribute，可选
payload
其中checksum采用CRC32算法计算，计算的内容包括消息属性长度+消息属性+data，消息属性如果不存在则不包括在内。消费者在接收到消息后会检查checksum是否正确。
　　同一个topic下有不同分区，每个分区下面会划分为多个文件，只有一个当前文件在写，其他文件只读。当写满一个文件（写满的意思是达到设定值）则切换文件，新建一个当前文件用来写，老的当前文件切换为只读。文件的命名以起始偏移量来命名。看一个例子，假设meta-test这个topic下的0-0分区可能有以下这些文件：
　　00000000000000000000000000000000.meta
00000000000000000000000000001024.meta
00000000000000000000000000002048.meta
……
其中00000000000000000000000000000000.meta表示最开始的文件，起始偏移量为0。第二个文件00000000000000000000000000001024.meta的起始偏移量为1024，同时表示它的前一个文件的大小为1024-0=1024。同样，第三个文件00000000000000000000000000002048.meta的起始偏移量为2048，表明00000000000000000000000000001024.meta的大小为2048-1024=1024。
　　以起始偏移量命名并排序这些文件，那么当消费者要抓取某个起始偏移量开始位置的数据变的相当简单，只要根据传上来的offset*二分查找*文件列表，定位到具体文件，然后将绝对offset减去文件的起始节点转化为相对offset，即可开始传输数据。例如，同样以上面的例子为例，假设消费者想抓取从1536开始的数据1M，则根据1536二分查找，定位到00000000000000000000000000001024.meta这个文件（1536在1024和2048之间），1536-1024=512，也就是实际传输的起始偏移量是在00000000000000000000000000001024.meta文件的512位置。因为1024.meta的大小才1K，比1M小多了，实际传输的数据只有2048-1536=512字节。
　　这些文件在meta里命名为Segment，每个Segment对应一个FileMessageSet。文件组织成SegmentList，整体成为一个MessageStore，一个topic下的一个分区对应一个MessageStore。
　　异步消息发送（发送消息后返回的结果中不包含准确的messageId,partition,offset,这些值都是-1）：
　　使用场景：
对于发送可靠性要求不那么高,但要求提高发送效率和降低对宿主应用的影响，提高宿主应用的稳定性.
例如,收集日志或用户行为信息等场景。

// New session factory,强烈建议使用单例 final AsyncMessageSessionFactory sessionFactory = newAsyncMetaMessageSessionFactory(initMetaConfig()); // create producer,强烈建议使用单例 final MessageProducer producer = sessionFactory.createAsyncProducer();

个人测试：
　　异步slave的情况下
　　master负责接送生产者的发送的数据，如果master故障 生产者的发送会报异常，但消息的消费还可以继续消费数据。

源码分析：

　　1：ZKLoadRebalanceListener负责发现master slave( log.info("Stopping fetch runners,maybe master or slave changed");) topice 或 partition 变化需要更新本地的old(按策略)
Cluster保存(ConcurrentHashMap<Integer/* broker id */, Set<Broker>> brokers(master(brokerid 1,slaveid 0)) slave(brokerid 1,slaveid 1))
brokerIdsPath/master
brokerIdsPath/slave(slaveId)
　　this.dirs.consumerRegistryDir（保存consumer）/
this.dirs.consumerRegistryDir/consumerId（保存topices）
　　this.brokerTopicsSubPath/topic（保存brokers）
this.brokerTopicsSubPath/topic/broker（保存partitionString（json））
　　

　　meta 1.4.3支持异步maste->slave