MetaQ文档学习

不念不忘少年蓝@ 2021-12-19 03:39 523阅读 0赞

# 消息中间件和RPC(HSF)异同 #

![在这里插入图片描述][20190702121629151.png]

## RPC适用场景 ##

*  调用时双向的
 *  调用方需要依赖多个服务提供结果

## 消息队列适用场景 ##

*  消息发送不受限于消息消费方处理速度
 *  发送方通过queue进行消息堆积，防止击穿下游服务
 *  多个下游服务可以订阅同一个topic进行处理
 *  处理耗时操作，比如文件复制，网络下载等等。

# 概念和术语 #

## 消息生产者 ##

简称producer，负责消息产生并发送到meta服务器

## 消息消费者 ##

简称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转化为具体文件的相对偏移量。详细内容参见\#消息的存储结构

# MetaQ架构 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]

## NameServer ##

服务的发现，维护了Broker的地址列表和Topic及Topic对应队列的地址列表，与每一个broker保持心跳连接，检车broker是否存活，在producer和consumer需要发布或者消费消息的时候，想nameserver发出请求获取连接。

## Broker ##

broker节点是MetaQ消息存储的位置，所有的消息都由节点broker负责存储。broker会与nameserver建立长连接，将自身的相关信息（比如ip、topic、分区数等）发送到nameserver中，由nameserver集群对节点集群的相应信息进行维护。

## Producer ##

生产者发消息前需要与nameserver建立长连接，生产者将从nameserver上获取到的相关broker节点的信息（比如broker上的topic类型，分区大小等）保存在本地。生产者根据自身需要发送的消息topic类型，从本地broker缓存文件中获取到节点列表，并从中选择某个broker节点，得到节点的相应address信息，然后生产者与这个节点建立连接。如果需要发送的topic在本地缓存中找不到对应的broker，生产者则会根据自身启动时初始化的remotingClient变量（记录了服务器的相关ip和端口），对nameserver进行请求。将自身所提供的topic信息注册到对应的broker节点上。如果注册成功，则最终会返回一个topicPublishInfo，其中记录了存有topic的对应broker列表信息。

生产者产生的消息体的主要由四部分构成：topic、tag、key、msgBody。

*  topic：消息的主题。发布和订阅消息是，都以topic为区分标准
 *  tag：可看做topic的二级分类。一般可以用来对消息进行过滤处理，以获取到更精确的消息。
 *  key：消息的唯一标识符。用于标识一个消息。当消息的传输和存储过程中出现故障问题时，可通过key来对失败的消息进行快速定位与查找，对于后期维护以及排查错误是一个关键的变量。
 *  msgBody：消息体。消息的主体部分一般是byte类型数组，在java开发过程中，往往需要通过序列化和压缩等方式对其进行组装。

## Consumer ##

同生产者一样，消费者需要与nameserver建立长连接，消费者将从nameserver上获取到的相关broker节点的信息（比如broker上的topic类型、节点数量、分区大小等）保存在本地。

消费者在进行消息获取时有两种方式：第一种是pull模型，第二种是push模型。

*  \*\*pull模型：\*\*即消费者主动与broker节点进行连接通信，然后根据自己所需要的topic消息类型，从broker上拉取下来指定数量的消息。主动权在消费者手中，pull方式的循环间隔不好设定，间隔设定太短，处在“忙等”状态，浪费资源，间隔太长，消息不能及时处理
 *  \*\*push模型：\*\*broker节点在收到生产者消息后，主动将消息推送到consumer上。Push方式的实时性比较高，但是会加大server端的工作压力，而且由于client的处理能力不能，client不能受server控制。

metaQ采用长轮询pull方式，既解决了pull实时性不够的问题，又不至于大量浪费资源。

# 消息存储 #

MetaQ的存储方式采用**物理队列+逻辑队列**的形式。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 1]

## 物理队列： ##

一台机器只有一个，也就是本地的文件系统（图中的commit log），存储着实际的数据文件。MetaQ将消息存储在本地文件中，每个文件最大大小为1G，如果写入新的消息时，超过当前文件大小，则会自动新建一个文件。文件名称为起始字节大小。以起始字节大小命名并排序这些文件是有诸多好处的，当消费者要抓取某个起始偏移量开始位置的数据，会变的很简单，只要根据传上来的offset二分查找文件列表，定位到具体文件，然后将绝对offset减去文件的起始节点转化为相对offset，即可找到对应的数据。

### 刷盘 ###

commit log是以append的方式编写的，保证了顺序写磁盘，顺序写磁盘效率比随机写内存还高，高吞吐量的保证

*  同步刷盘：节点收到消息之后，会立刻把消息写入到磁盘中
 *  异步刷盘：节点收到消息之后，并不会第一时间把消息写到磁盘中，而是先写到内存中。当收到N条消息或者经过一段时间后会在统一把消息写入到磁盘中。

### 读盘 ###

metaQ的所有消息都是持久化的，先写入系统的PageCache(页高速缓存)，然后刷盘，可以保证内存和磁盘都有一份数据，访问时，直接从内存中获取。

\[外链图片转存失败(img-fuGJtODu-1562040940329)(C:\\Users\\xiaozhan.fc\\AppData\\Roaming\\Typora\\typora-user-images\\1561026231679.png)\]

## 逻辑队列 ##

一台机器可以有多个(topicA\_3 topicA-4 等等)，逻辑队列中的存储的是索引文件。服务器将消息存储到文件后，会将该消息在文件的物理位置，消息大小，消息类型封装成一个固定大小的数据结构，暂且称这个数据结构为索引单元吧，大小固定为 16 byte，消息在物理文件的位置称为offset，8个字节，消息size占4个字节，MessageType占4个字节。多个索引单元组成了一个索引文件，索引文件默认固定大小为 20M，和消息文件一样，文件名是起始字节位置，写满后，产生一个新的文件。metaq对于客户端展现的是逻辑队列就是消费队列，consumer从消费队列里顺序取消息进行消费。

这种设计是把物理和逻辑分离，消费队列更加轻量化。所以metaq可以支撑更多的消费队列数，提升消息的吞吐量，并且有一定的消息堆积能力。但是也有缺点：虽然是顺序写入，但是读却是随机读的。

# 可靠性 顺序 重复 #

## 可靠性 ##

### 生产者可靠性 ###

消息生产者发送消息后返回SendResult，如果isSuccess返回true，则表示消息已经确认发送到服务器，并且被服务器接受存储。整个发送过程是一个同步的过程，保证消息送到服务器并且返回结果。

### 服务器可靠性 ###

1.  收到消息之后，写入磁盘，写入成功之后，返回应答给生产者
2.  os对系统有缓冲：
    
     *  每1000条（可配置），即强制调用一次force来写入磁盘设备。
     *  每隔10秒（可配置），强制调用一次force来写入磁盘设备。

因此,Meta通过配置可保证在异常情况下（如磁盘掉电）10秒内最多丢失1000条消息。当然通过参数调整你甚至可以在掉电情况下不丢失任何消息。

### 消费者可靠性保证 ###

消费者是一条一条消费信息。

如果消费某条信息失败（如异常），则会尝试重试消费这条消息（默认最大次数5次）

超过最大次数无法消费，则将消息存储在本地磁盘，由后台线程继续重试。主线程往后走，继续消费。只有在MessageListener确认成功消费一条消息后，meta的消费者才会继续消费另一条消息。由此来保证消息的可靠消费。

**存储方案，offset存储**

*  zookeeper，默认存储在zoopkeeper上，zookeeper通过集群来保证数据安全性
 *  mysql，可以连接到使用的mysql数据库，只要建立一张特定的表来存储。完全由数据库来保证数据可靠性
 *  file，文件存储，将offset信息存储在消费者的本地文件中。

## 顺序 ##

默认处理原则：谁先到达服务器并写入磁盘，则先处理谁。因为消费者针对每个分区都是按照从前到后递增offset的顺序拉取消息。

Meta可以保证，在单线程内使用该producer发送的消息按照发送的顺序达到服务器并存储，并按照相同顺序被消费者消费，前提是这些消息发往**同一台服务器的同一个分区**。

public interface PartitionSelector { 
        public Partition getPartition(String topic, List<Partition> partitions, Message message) throws MetaClientException;
    }

## 消息重复 ##

消息重复发生的例子：

生产者：生产者发送消息，等待服务器应答，这个时候发生网络故障，服务器实际上已经将消息写入成功，但是由于网络故障没有返回应答，那么生产者会**重发**，这个时候服务器收到了两条相同的信息。

这种由故障引起的重复，meta是无法避免的，不判断消息的data是否一致，meta仅仅作为载荷来传输

针对消费者来说也会有这个问题，消费的时候，机器突然断电，没有及时将前进后的offset存储起来，则下次启动的时候或者其他同个分组的消费者owner到这个分区的时候，会重复消费这条消息。

# 消息生产者(Producer) #

首先会用到消息回话工厂类——MessageSessionFactory，这个工厂作用是创建生产者和消费者。创建生产者代码：

final MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(initMetaConfig());
            // create producer,强烈建议使用单例
    final MessageProducer producer = sessionFactory.createProducer();

点进去 MetaMessageSessionFactory之后，看看这个工厂帮助我们做了哪些事情。

public MetaMessageSessionFactory(final MetaClientConfig metaClientConfig) throws MetaClientException { 
            super();
            try { 
                this.checkConfig(metaClientConfig);
                this.metaClientConfig = metaClientConfig;
                final ClientConfig clientConfig = new ClientConfig();
                clientConfig.setTcpNoDelay(TCP_NO_DELAY);
                clientConfig.setMaxReconnectTimes(MAX_RECONNECT_TIMES);
                clientConfig.setWireFormatType(new MetamorphosisWireFormatType());
                clientConfig.setMaxScheduleWrittenBytes(MAX_SCHEDULE_WRITTEN_BYTES);
                try { 
                    this.remotingClient = new RemotingClientWrapper(RemotingFactory.connect(clientConfig));
                }
                catch (final NotifyRemotingException e) { 
                    throw new NetworkException("Create remoting client failed", e);
                }
                // 如果有设置，则使用设置的url并连接，否则使用zk发现服务器
                if (this.metaClientConfig.getServerUrl() != null) { 
                    this.connectServer(this.metaClientConfig);
                }
                else { 
                    this.initZooKeeper();
                }
    
                this.producerZooKeeper =
                        new ProducerZooKeeper(this.metaZookeeper, this.remotingClient, this.zkClient, metaClientConfig);
                this.sessionIdGenerator = new IdGenerator();
                // modify by wuhua
                this.consumerZooKeeper = this.initConsumerZooKeeper(this.remotingClient, this.zkClient, this.zkConfig);
                this.zkClientChangedListeners.add(this.producerZooKeeper);
                this.zkClientChangedListeners.add(this.consumerZooKeeper);
                this.subscribeInfoManager = new SubscribeInfoManager();
                this.recoverManager = new RecoverStorageManager(this.metaClientConfig, this.subscribeInfoManager);
                this.shutdownHook = new Thread() { 
    
                    @Override
                    public void run() { 
                        try { 
                            MetaMessageSessionFactory.this.isHutdownHookCalled = true;
                            MetaMessageSessionFactory.this.shutdown();
                        }
                        catch (final MetaClientException e) { 
                            log.error("关闭session factory失败", e);
                        }
                    }
    
                };
                Runtime.getRuntime().addShutdownHook(this.shutdownHook);
            }
            catch (MetaClientException e) { 
                this.shutdown();
                throw e;
            }
            catch (Exception e) { 
                this.shutdown();
                throw new MetaClientException("Construct message session factory failed.", e);
            }
        }

可以看到，这个工厂大致帮我们做了下面几个事情：

1.  服务的查找和发现，通过diamond和zookeeper帮你查找日常的meta服务器地址列表
2.  连接的创建和销毁，自动创建和销毁到meta服务器的连接，并做连接复用，也就是到同一台meta的服务器在一个工厂内只维持一个连接。
3.  消息消费者的消息存储和恢复。
4.  协调和管理各种资源，包括创建的生产者和消费者的。

# 消息 #

## 消息属性 ##

消息是邮件，MetaQ是邮局，消息填好目的地之后，放在邮局，邮局就可以帮助你送到目的地点  
MetaQ的消息在Java客户端里是com.taobao.metamorphosis.Message类，它主要包括这么几个属性：

public class Message implements Serializable { 
        static final long serialVersionUID = -1L;
        private long id;
        private String topic;
        private byte[] data;
        private String attribute;
        private int flag;
        private Partition partition;
        private transient boolean readOnly;
        private transient boolean rollbackOnly = false;

*  id：id 一个64位的消息ID，它在同一个MetaQ集群内唯一并且单调递增。也就是说在同一个线程内发送的消息，他们的消息id是顺序递增的。这个id不可设置，它是在服务端帮你自动生成，你只能读取这个id，通过getId方法。通常，我们可以利用这个唯一的id来做消息去重。
 *  topic：消息的主题
 *  paitition：消息的分区
 *  data：消息的载体
 *  attribute：消息的附加属性，一个字符串，可有可无。可以通过`hasAttribute`来检测这个消息是否有附加属性。
 *  readOnly 消息是否为只读，可将消息设置为只读状态（通过setReadOnly(true))，只允许读，不可以更改。
 *  ollback 消息是否回滚。这只对消费者有意义，消费者可以主动回滚一个消息，通过setRollbackOnly(true)。那么这个消息将会被重新消费。

int getFlag() { 
            return this.flag;
        }
    
        void setFlag(int flag) { 
            this.flag = flag;
        }
    
        boolean isRollbackOnly() { 
            return this.rollbackOnly;
        }
    
        public void setRollbackOnly() { 
            this.rollbackOnly = true;
        }
    
        public boolean isReadOnly() { 
            return this.readOnly;
        }
    
        public void setReadOnly(boolean readOnly) { 
            this.readOnly = readOnly;
        }
    
        public boolean hasAttribute() { 
            return this.attribute != null;
        }

## 创建消息 ##

Message有两种构造方法：

public Message(String topic, byte[] data) { 
            this.topic = topic;
            this.data = data;
        }
    
        public Message(String topic, byte[] data, String attribute) { 
            this.topic = topic;
            this.data = data;
            this.attribute = attribute;
        }

### 创建不带附加属性的消息 ###

final String topic = "meta-test";
    final byte [] data = "hello world".getBytes();
    final Message msg = new Message(topic, data);

getBytes()：使用平台的默认字符集将字符串编码为 byte 序列，并将结果存储到一个新的 byte 数组中。

### 创建带属性的消息 ###

final String topic = "meta-test";
    final byte [] data = "hello world".getBytes();
    final String attribute = "attribute";
    final Message msg = new Message(topic, data, attribute);
    System.out.println(msg.hasAttribute()); //true

附加属性可以用作分区选择或者消息过滤

### 只读消息 ###

msg.setReadOnly(true);
    System.out.println(msg.isReadOnly()); //true
    msg.setTopic(newTopic);  //throw IllegalStateException.
    msg.setReadOnly(false);
    msg.setTopic(newTopic); //set new topic successfully.

将消息设置为只读后，再尝试修改消息的topic或者data，都会抛出IllegalStateException异常。一般你会将消息设置为只读来防止消息的意外修改。

### 发送之后获取id和分区 ###

final MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(initMetaConfig());
    final MessageProducer producer = sessionFactory.createProducer();
    SendResult rt = producer.send(msg);
    long id = msg.getId();
    Partition part = msg.getPartition();
    int brokerId = part.getBrokerId();
    int partition = part.getPartition();

发送后，`MessageProducer`会自动地将服务端返回的id和分区信息填入message。  
![在这里插入图片描述][20190702121802610.png]

# 配置客户端 #

在创建生产者和消费者之前，都需要配置客户端，客户端的配置是通过`com.taobao.metamorphosis.client.MetaClientConfig`类来进行配置的。

public class MetaClientConfig implements Serializable { 
      protected ZKConfig zkConfig;
      public ZKConfig getZkConfig() { 
            return this.zkConfig;
        }
    }

zkConfig是com.taobao.metamorphosis.utils.ZkUtils.ZKConfig类，它包括6个属性，源代码如下:

@Key(name = "zk.zkRoot")
            public String zkRoot = "/meta";
            /** * If enable zookeeper */=
            @Key(name = "zk.zkEnable")
            public boolean zkEnable = true;
    
            /** ZK host string */
            @Key(name = "zk.zkConnect")
            public String zkConnect;
    
            /** zookeeper session timeout */
            @Key(name = "zk.zkSessionTimeoutMs")
            public int zkSessionTimeoutMs = 30000;
    
            /** * the max time that the client waits to establish a connection to * zookeeper */
            @Key(name = "zk.zkConnectionTimeoutMs")
            public int zkConnectionTimeoutMs = 30000;
    
            /** how far a ZK follower can be behind a ZK leader */
            @Key(name = "zk.zkSyncTimeMs")
            public int zkSyncTimeMs = 5000;

*  zkconfig：zookeeper配置，MetaQ是通过zookeeper来做服务和路由查找。服务端将自己连接到zookeeper，而客户端连接到zookeeper之后，查找服务端的信息，然后再跟服务端打交道
 *  zkRoot， MetaQ服务端集群的根path，必须跟服务端配置保持一致，这就类似电话本里的姓名，姓名不对，你就找不到正确的电话号码。zkRoot不对，你也找不到正确的MetaQ服务端集群。
 *  zkConnect， zookeeper集群地址列表，形如host:port,host:port的字符串。
 *  zkSessionTimeoutMs, zookeeper客户端连接的最大超时时间，默认为30秒，单位毫秒.
 *  zkConnectionTimeoutMs, 连接zk集群的最大超时时间，默认也是30秒，单位毫秒。
 *  zkSyncTimeMs， zookeeper集群内的数据同步最大延迟，这个需要你去测量和估计，默认是5秒，单位毫秒。消费者在负载均衡失败后，会至少等待这么长时间再尝试。  
    配置代码：

final MetaClientConfig metaClientConfig = new MetaClientConfig();
     final ZKConfig zkConfig = new ZKConfig();
     zkConfig.zkConnect = "127.0.0.1:2181";
     zkConfig.zkRoot = "/meta";
     metaClientConfig.setZkConfig(zkConfig);

# 会话工厂MessageSessionFactory #

## 介绍 ##

创建producer或者consumer之前都必须先创建一个MessageSessionFactory实例，作为工厂才能具体的生产者和消费者。通常，除非你拥有多个MetaQ集群，否则在你的应用里一般也只会拥有一个单例的消息会话工厂。

## 创建工厂 ##

创建工厂在客户端配置之后：

final MetaClientConfig metaClientConfig = new MetaClientConfig();
    final ZKConfig zkConfig = new ZKConfig();
    zkConfig.zkConnect = "127.0.0.1:2181";
    metaClientConfig.setZkConfig(zkConfig);
    
    MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);

或者：

final MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(initMetaConfig());

## 关闭工厂 ##

创建的消息会话工厂，必须明确地调用shutdown方法才能关闭，否则会有资源泄漏的隐患。

# 消息生产者MessageProducer #

MQ系统的三个角色：生产者，消费者和队列。生产者作为消息内容的提供商，是整个系统的驱动者。生产者发送消息到队列服务器，消费者从队列服务器获取消息并消费。三者分工明确，密切协作来完成异步任务。

## 创建生产者 ##

就可以通过工厂来创建消息生产者，这都是通过createProducer方法：

final MessageProducer producer = sessionFactory.createProducer();

**但是这样还是不能够马上发送消息**  
客户端链接到zookeeper后还不知道应该连接哪个服务器。它必须告诉zookeeper想去连接哪个topic的服务器，拿到服务器地址，然后才能建立socket连接，最后才可以发送消息。  
因此创建生产者之后还需要一个发布Topic的过程：、

// publish topic
            final String topic = "meta-test";
            producer.publish(topic);

假设我们要发送的消息的主题是meta-test，那么只要调用publish(topic)这个方法就可以告诉zookeeper我们想要查找提供topic为meta-test的服务器，从zookeeper获取服务器地址后，会话工厂会连接服务器，生产者就可以使用这些连接来发送消息了。  
publish(topic)方法可以针对同一个topic调用多次，这跟调用一次的效果是一样的。

同一个生产者想要发送多个不同topic的消息，那么这些topic都需要被发布一次:

producer.publish("topic1");
        producer.publish("topic2");

例如，这里我们发布了两个topic：topic1和topic2，接下来就可以尝试发送这些主题的消息。

## 在zookeeper服务器里配置topic ##

我们发布了meta-test的主题，然后希望zookeeper告诉我们哪里有提供meta-test主题的队列服务的服务器，但是zookeeper可能找不到提供这个主题服务的服务器，那么在发送消息的时候就会看到这样的异常： There is no aviable partition for topic meta-test,maybe you don’t publish it at first?，这种情况你必须检查你的服务器是否配置了meta-test这个topic，并且客户端和服务器的zookeeper配置也保持一致。  
配置topic可以在conf里面配置。

## 发送消息 ##

使用send方法发送消息：

// send message
             try{ 
                final SendResult sendResult = producer.sendMessage(new Message(topic, "hello,MetaQ!".getBytes()));
                // check result
                if (!sendResult.isSuccess()) { 
                    System.err.println("Send message failed,error message:" + sendResult.getErrorMessage());
                }
                else { 
                    System.out.println("Send message successfully,sent to " + sendResult.getPartition());
                }
             }catch(MetaClientException e){ 
                 e.printStackTrace(); //TODO log
             }catch(InterruptedException e){ 
                 Thread.currentThread().interrupt();
             }

返回SendResult实例。  
\- SendResult.isSuccess 返回一个布尔值可以判断发送成功或者失败、  
\- SendResult.getErrorMessage() 可以看到出错信息  
\- 如果发送成功，则可以通过getPartition()和getOffset()方法获取消息被发送到哪个分区，以及消息在该分区数据文件中的绝对偏移量。  
\- send(message)是一个同步调用，默认使用3秒的超时时间，如果超时还没有返回则会抛出MetaOpeartionTimeoutException异常，这个异常是MetaClientException异常的子类。如果要捕捉超时异常，可以单独catch这个异常信息。  
\- send同时有一个同步的重载的方法用于设定发送超时时间，比如5秒发送超时`producer.send(message, 5000, TimeUnit.MILLISECONDS);`

## 获得分区方法 ##

//第一种获得分区的方法
            long id = msg.getId();
            Partition part = msg.getPartition();
            int brokerId = part.getBrokerId();
            int partition = part.getPartition();
            System.out.println("id : " + id + " part : " + part + " brokerId : " + brokerId + " pritition : " + partition);
    
            //另外一种获得分区的方法
            Partition part1 = sendResult.getPartition();
            int brokerId1 = part1.getBrokerId();
            int partition1 = part1.getPartition();
            System.out.println(" brokerId : " + brokerId + " pritition : " + partition);

1.  第一种方法是消息发送之后，从消息中获取
2.  第二种方法是消息发送之后，从消息结果中获取

## 异步发送消息 ##

同步发送消息会阻塞当前线程，直到服务端返回发送结果之后，才能够取消阻塞，MessageProducer提供异步的消息发送机制。  
同步消息发送出去之后，会有发送结果返回，但是异步发送出去之后，不需要返回值。

public class AsynchronousProducer { 
        public static void main(String[] args) throws Exception{ 
            //客户端配置
            final MetaClientConfig metaClientConfig = new MetaClientConfig();
            final ZKConfig zkConfig = new ZKConfig();
            zkConfig.zkConnect = "127.0.0.1:2181";
            metaClientConfig.setZkConfig(zkConfig);
    
            //创建会话工厂
            final MessageSessionFactory sessionFactory = new MetaMessageSessionFactory(metaClientConfig);
            final MessageProducer producer = sessionFactory.createProducer();
    
            //发布topic
            final String topic = "mata-test";
            producer.publish(topic);
    
            //异步发送消息
            producer.sendMessage(new Message(topic, "hello MetaQ!".getBytes()), new SendMessageCallback() { 
                public void onMessageSent(SendResult sendResult) { 
                    if(sendResult.isSuccess()){ 
                        System.out.println("Send message successfully, send to " + sendResult.getPartition());
                    }
                    else { 
                        System.err.println("Send Message Failed, error message" + sendResult.getErrorMessage());
                    }
                }
    
                public void onException(Throwable throwable) { 
                    throwable.getStackTrace();
                }
            });
    
    
        }
    }

采用回调函数来判断正误。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 2]  
可以看到，结果显示发送成功，并且是采用自己配置的集群地址列表：127.0.0.1  
send(msg, callback)除了接收消息外，还接收一个实现了接口SendMessageCallback的回调对象。这个接口有两个必须实现的方法：

*  onMessageSent(sendResult)，当消息发送后，返回发送的结果，检查sendResult来判断成功还是失败。
 *  onException(throwable)，当客户端异常发生的时候，此时消息已经发送失败，你可以处理这个异常或者简单地打印日志记录。

异步的send方法就没有返回结果了，发送结果通过onMessageSent返回给回调对象。同样，异步的send方法也有指定超时的重载版本send(message, callback, timeout, timeUnit)

## 异步发送流的监控 ##

异步发送消息的时候，在发送结果从服务端返回之前，传入的callback对象会一直存放在内存中，由于异步的send方法不会阻塞，因此理论上发送你不断的发送异步消息，就注册无穷多个callback到生产者那里等待通知。但是由于内存有限，很可能过多的callback对象导致内存爆满而引发频繁的Full GC，导致服务彻底不可用。因此异步发送应该做一定的流量控制限制，比如这里提供一个简单的例子，使用信号量做流控：

//最多同时存在1万个回调消息
         Semaphore permits= new Semaphore(10000);
            while (true) { 
                // 每次发送前都请求许可
                if (permits.tryAcquire()) { 
                    // 获取许可成功，发送消息
                    try { 
                        producer.send(message, new SendMessageCallback() { 
                            @Override
                            public void onMessageSent(SendResult result) { 
                                try { 
                                    // 处理发送结果result
                                }
                                finally { 
                                    // 切记释放许可
                                    permits.release();
                                }
                            }
                            @Override
                            public void onException(Throwable e) { 
                                e.printStackTrace();
                                // 切记释放许可
                                permits.release();
                            }
                        });
                    }
                    catch (Exception e) { 
                        e.printStackTrace();
                    }
                    //已发送，跳出循环。
                    break;
                }
                else { 
                    // 让出执行权，等待再次获取许可
                    Thread.yield();
                }
            }

我们设置信号量的最大许可为1万，然后每次发送消息前都尝试申请许可，申请成功才可以发送消息，然后在callback里释放许可（无论是发送成功失败还是产生异常都需要释放许可）；如果申请失败，则调用yield让出执行权，等待while循环的下次重试。

## 发送事务消息 ##

MetaQ支持发送事务消息，包括本地事务和支持XA协议（二阶段提交协议）的分布式事务。

### 1. 发送本地事务消息 ###

所谓本地事务消息，通俗来说就是批量发送一批消息，要么同时发送成功，要么同时发送失败，保证事务的ACID的特性。MessageProducer类有beginTransaction()，commit()和rollback()方法用来控制一个本地事务，一个具体的例子：

// 开始事务
    producer.beginTransaction();
    try{ 
      SendResult result1 = producer.send(msg1);
      SendResult result2 = producer.send(msg2);
    
      if(result1.isSuccess() && result2.isSuccess()){ 
    	producer.commit();
      }else{ 
    	//任何一条发送失败都应该回滚消息
    	producer.rollback();
      }
    
    }catch(Exception e){ 
      //发生异常，回滚消息
      producer.rollback();
    }

通过producer.beginTransaction开启一个本地的事务，然后在事务里面发送两条消息，只有在两条消息都发送成功的时候才能通过commit方法提交事务，否则都使用rollback回滚事务，同样，在发生异常的时候也应该回滚事务。

事务还可以设置超时时间：

//设置事务超时为10秒
     producer.setTransactionTimeout(10);

超过这个时间还没有提交或者回滚的事务，将被服务器自动回滚。事务的最大超时时间不能超过服务器设置的maxTxTimeoutInSeconds，超过会被设置为maxTxTimeoutInSeconds设定的最大值。  
beginTransaction,rollback和commit本质上是跟服务器做通讯，告诉服务器开始，回滚或者提交某个事务，通讯必然涉及到通讯超时的问题，也可以设置这些事务请求的超时时间，默认为5秒：

//设置事务请求超时为10秒。
    producer.setTransactionRequestTimeout(10000,TimeUnit.MILLISECONDS);

通常来说这个值不应该超过事务的时间时间。

### 2. 发送XA事务消息 ###

所谓XA事务消息，是指在一个事务内除了MetaQ这个事务源之外，还有另外一个事务源参与了事务，最常见的比如数据库源。一个典型的场景是：往MetaQ发送消息，同时要向数据库插入一条数据，两个操作要么同时成功，要么同时失败。不允许出现发送消息成功，而插入数据库失败的情况，反之亦然。比如下订单这个操作，要往订单表插入一条记录，同时发送一条消息到MetaQ，执行一些异步任务如通知用户、物流，记录日志，统计分析等等，就需要分布式事务。  
假设有一个订单表

CREATE TABLE `test`.`orders` (
    	`id` int(11) NOT NULL AUTO_INCREMENT,
    	`uid` int(11) NOT NULL,
    	`address` varchar(255) NOT NULL,
    	PRIMARY KEY (`id`)
    );

现在我们就要模拟一个下订单的操作，同时插入一条记录到订单表，并且发送一条消息给MetaQ处理。  
首先，创建XAMessageSessionFactory实例，跟创建MessageSessionFactory类似：

final XAMessageSessionFactory xasf = new XAMetaMessageSessionFactory(metaClientConfig);

接下来创建XAMessageProducer并发布topic:

final String topic = "meta-test";
     XAMessageProducer xaMessageProducer = xasf.createXAProducer();
     // publish topic
     xaMessageProducer.publish(topic);

XAMessageProducer就是一个参与分布式事务的事务源，接下来要创建一个XADataSource来代表数据库事务源，注意必须使用数据库的XA数据源：

final MysqlXADataSource mysqlXADataSource = new MysqlXADataSource();
           mysqlXADataSource
           .setUrl("jdbc:mysql://localhost:3306/test?characterEncoding=utf8&connectTimeout=1000&autoReconnect=true");
           mysqlXADataSource.setUser("root");
           mysqlXADataSource.setPassword("");
           mysqlXADataSource.setPreparedStatementCacheSize(20);

## 发送默认topic消息 ##

MetaQ的topic必须配置，否则汇报异常，  
但是常常会有这样的需求，希望能发送任何类型topic的消息，而不是每次都要在MetaQ服务端配置topic。这可以通过设置MessageProducer的默认topic来实现。假设你发送topic1的消息，当生产者在zookeeper上找不到提供topic1的服务器，那就会选择提供默认topic的服务器来发送这条消息。  
首先在服务端配置一个默认topic，如：

[topic=app-*]

接下来设置MessageProducer的默认topic:

producer.setDefaultTopic("app-*");

接下来你就可以发布任意以app-开头的topic的消息，例如：

producer.send(new Message("app-topic1","hello world".getBytes()));
    producer.send(new Message("app-topic2","hello world".getBytes()));
    producer.send(new Message("app-topic3","hello world".getBytes()));

哪怕这些topic没有被publish过。这些topic都将在服务端动态创建，无需明确配置。

我们这里以app-*作为默认topic，如果你不想有任何前缀设置，那也可以使用*作为默认topic，这样就可以发送任意类型topic的消息。

# 消息消费者MessageConsumer #

在消息发送者发送消息成功后，，消息就存储在服务端的数据目录里面（服务端datapath配置的目录），按照topic-partition的格式存储。接下来，消费者就要消费该消息。

## 配置消费者 ##

每个Java的消费者都需要一个ConsumerConfig的配置实例。

## 消费者分组 ##

在MetaQ里，消费者被认为是一个集群，也就是说认为是有一组的机器在共同分担消费一个topic。因此消费者配置ConsumerConfig中最重要的配置是group，每个消费者都必须告诉MetaQ它属于哪个group，然后MetaQ会找出这个group下所有注册上来的消费者，在他们之间做负载均衡，共同消费一个或多个topic。注意，不同group之间可以认为是不同的消费者，他们消费同一个topic下的消息的进度是不同。

举例来说，假设你有一个topic为business-logs，是所有业务系统的日志。然后现在你对这些日志要做两个事情：一个是存储到HDFS这样的分布式文件系统，以便后续做分析处理；以个是Twitter Storm这样的实时分析系统，做实时的数据分析、告警和展现。显然，这里你就需要两个group，比如我们有一个group叫hdfs-writer，它有三台机器同时消费business-logs，将日志存储到HDFS集群。同时，你也有另一个group叫storm-spouts，有5台机器用来给storm集群喂数据。这两个group是隔离，虽然是消费同一个topic，但是两者是消费进度（消费了多少个消息，等待消费多少个消息等信息）是不同的。但是同一个group内，例如hdfs-writer的三台机器，这三台机器是共同消费business-logs下的消息，同一条消息只会被这hdfs-writer三台机器中的一台处理，但是这条消息还会被twitter-spouts等其他分组内的某一台机器消费。

## 创建ConsumerConfig ##

创建ConsumerConfig并传入分组名称：

final String group = "hdfs-writer";
    ConsumerConfig consumerConfig = new ConsumerConfig(group);

ConsumerConfig定义，继承自MetaClientConfig

public class ConsumerConfig extends MetaClientConfig { 
        static final long serialVersionUID = -1L;
        private int fetchRunnerCount = Runtime.getRuntime().availableProcessors();
        private long maxDelayFetchTimeInMills = 5000L;
        /** @deprecated */
        @Deprecated
        private long maxDelayFetchTimeWhenExceptionInMills = 10000L;
        private long fetchTimeoutInMills = 10000L;
        private String consumerId;
        private String partition;
        private long offset = 0L;
        private String group;
        private long commitOffsetPeriodInMills = 5000L;
        private int maxFetchRetries = 5;
        private boolean alwaysConsumeFromMaxOffset = false;
        private Type loadBalanceStrategyType;
        private int maxIncreaseFetchDataRetries;

ConsumerConfig的其他重要选项还包括：

*  fetchRunnerCount, 因为MetaQ的消费者是以pull模型来从服务端拉取数据并消费，这个参数设置并行拉取的线程数，默认是CPUs个。
 *  fetchTimeoutInMills，同步抓取的请求超时，默认10秒，通常不需要修改此参数。
 *  maxDelayFetchTimeInMills，当上一次没有抓取到的消息，抓取线程sleep的最大时间，默认5秒，单位毫秒。当某一次没有抓取到消息的时候，抓取线程会开始休眠maxDelayFetchTimeInMills的10分之1时间，如果下次还是没有抓到，则休眠maxDelayFetchTimeInMills的10分之2时间，以此类推直到最多休眠maxDelayFetchTimeInMills时间。中途如果任何一次抓取开始获取数据，则计数清零从10分之1重新开始计算。当你对消息的实时性特别敏感的时候应该调小此参数，并同时调小服务端的unflushInterval参数。
 *  consumerId, 单个消费者的id，必须全局唯一，通常用于标识分组内的单个消费者，可不设置，系统会根据IP和时间戳自动生成。
 *  offset， 第一次消费开始位置的offset，默认都是从服务端的最早数据开始消费。
 *  commitOffsetPeriodInMills， 保存消费者已经消费的数据的offset的间隔时间，默认5秒，单位毫秒。更大的间隔，在故障和重启时间可能重复消费的消息更多，更小的间隔，可能给存储造成压力。
 *  maxFetchRetries，同一条消息在处理失败情况下最大重试消费次数，默认5次，超过就跳过这条消息并调用RejectConsumptionHandler处理。关于RejectConsumptionHandler请看下面的拒绝处理小节。

这些参数都有相应的getter/setter方法来设置。

## 创建消费者 ##

通过MessageSessionFactory的createConsumer方法即可创建消费者：

final MessageConsumer consumer = sessionFactory.createConsumer(consumerConfig);

需要传入的参数类型是MetaClientConfig，而consumerConfig是继承自MetaClientConfig的。

## offset的存储 ##

MetaQ的消费模型是一种拉取的模型，消费者根据上次消费数据的绝对偏移量(offset)从服务端的数据文件中拉取后面的数据继续消费，因此这个offset信息就非常关键，需要可靠地保存。默认情况下，MetaQ是将offset信息保存在你使用的zookeeper集群上，也就是ZkOffsetStorage所做的事情，它实现了OffsetStorage接口。通常这样的保存是可靠并且安全的，但是有时候可能你也需要其他选项，目前还提供两个不同的OffsetStorage实现：

*  LocalOffsetStorage，使用consumer的本地文件作为offset存储，默认存储在$\{HOME\}/.meta\_offsets的文件里。适合消费者分组只有一个消费者的情况，无需共享offset信息。例如广播类型的消费者就特别合适。
 *  MysqlOffsetStorage，使用Mysql作为offset存储，使用前需要创建表结构：

CREATE TABLE `meta_topic_partition_group_offset` (
      `id` int(11) NOT NULL AUTO_INCREMENT,
      `topic` varchar(255) NOT NULL,
      `partition` varchar(255) NOT NULL,
      `group_id` varchar(255) NOT NULL,
      `offset` int(11) NOT NULL,
      `msg_id` int(11) NOT NULL,
      PRIMARY KEY (`id`),
      KEY `TOPIC_PART_GRP_IDX` (`topic`,`partition`,`group_id`)
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8;

你也可以实现自己的OffsetStorage存储。如果你想使用除了zookeeper之外的offset存储，可以在创建消费者的时候传入：

final MessageConsumer consumer = sessionFactory.createConsumer(consumerConfig,new MysqlOffsetStorage(dataSource));

第一个参数是消费者配置，第二个参数是offset的存储方式。

### 第一次消费的offset初始值 ###

前面提到ConsumerConfig有个offset参数可以设置第一次消费的时候开始的绝对偏移量，默认这个参数是0，也就是从服务端现有消息的最小偏移量开始，从头开始消费所有消息。

但是，通常情况下，新的消费分组都是希望从最新的消息开始消费，ComsumerConfig提供了一个setConsumeFromMaxOffset(boolean always)方法来设置从最新位置开始消费。其中always参数表示是否每次消费者启动都从最新位置开始消费，这样就忽略了在消费者停止期间的消息。通常仅在测试的时候将always参数设置为true，以便每次测试最新的消息。除非你真的不需要消费者停止期间（比如重启间隔）的消息，否则不要将always设置为真。

## 订阅topic ##

创建了消费者之后，需要订阅某些topic进行处理  
每个想消费的topic都必须提供一个消息处理器，用来处理该topic下的消息，这跟JMS里的MessgaeListener概念是一样的。MessageListener接口如下：

/** * 异步消息监听器 * * @author boyan * @Date 2011-4-23 * */
    public interface MessageListener { 
        /** * 接收到消息，只有messages不为空并且不为null的情况下会触发此方法 * * @param messages * TODO 拼写错误，应该是单数，暂时将错就错吧 */
        public void recieveMessages(Message message) throws InterruptedException;
    
    
        /** * 处理消息的线程池 * * @return */
        public Executor getExecutor();
    }

*  recieveMessages接收到服务端的消息并做业务处理。
 *  getExecutor用于提供给consumer处理这些消费使用的线程池，可以为null。

## 订阅多个topic ##

consumer.subscribe("topic1",1024,messageListener1);
    consumer.subscribe("topic2",1024,messageListener1);
    consumer.subscribe("topic3",1024,messageListener2);

## 处理消息 ##

在recieveMessages处理每条你收到的消息，比如我可以简单地打印：

new MessageListener() { 
    
                @Override
                public void recieveMessages(final Message message) { 
                    System.out.println("Receive message " + new String(message.getData()));
                }
                @Override
                public Executor getExecutor() { 
                    return null;
                }
            }

设置完Listener之后，就可以消费消息

consumer.subscribe(topic, 1024 * 1024, new MessageListener() { 
    
                @Override
                public void recieveMessages(final Message message) { 
                    System.out.println("Receive message " + new String(message.getData()));
                }
    
    
                @Override
                public Executor getExecutor() { 
                    // Thread pool to process messages,maybe null.
                    return null;
                }
            });
            // complete subscribe
            consumer.completeSubscribe();

### 并发处理 ###

前面提到ConsumerConfig中的fetchRunnerCount用来设定从服务端拉取数据的并发线程数目，默认是CPUs个  
需要注意的是

1.  fetchRunnerCount是用来设置从服务端拉取数据的并发线程数目的。
2.  getExecutor是用来设置处理消息并发线程数据的。

如果你的MessageListener的getExecutor方法返回null，也就是不提供一个线程池来处理消费，那么这些消息就会在抓取线程上处理，因此在多核机器上，这些消息其实也是并发被处理的，在单核机器上就是串行处理了（因为默认只有一个抓取线程）。

但是当getExecutor返回一个线程池的时候，抓取线程将只负责抓取并解析出消息，消息的处理将交给你提供的线程池来处理。比如你可以设定并发30个线程来处理消息：

final ExecutorService executor = Executors.newFixedThreadPool(30);
    ......
    
    public Executor getExecutor(){ 
        return this.executor;
    }

**注意，getExecutor方法不要每次重复创建线程池，应当一直返回同一个线程池。**  
总结：

*  当getExecutor返回null，处理消息是在抓取线程上执行(fetchRunnerCount)，默认抓取线程数目等于你的CPU个数。如果你是多核机器，处理消息将是并发执行。
 *  当getExecutor返回一个线程池，消息处理将采用该线程池

### 线程池繁忙 ###

当线程池繁忙并且线程池的队列满的情况下，你可以采用不同的策略，比如默认的AbortPolicy，无论你采用什么策略，MetaQ的consumer都能保证只在消息被成功处理的情况下递增offset。

## 订阅和完成订阅 ##

订阅消息通过subscribe方法即可：

final String topic ="meta-test";
    int maxSize = 1024*1024;
    MessageListener messageListener=......
    consumer.subscribe(topic, maxSize, messageListener);

maxSize参数解释：因为MetaQ采用pull模型拉取消息消费，那么就涉及到每次拉取多少消息的问题。这个maxSize就是用来设置每一次拉取请求的buffer大小，单位是字节数。例如上面的例子是每次尽量拉取最多1M的消息。因为可能没有足够的消息来填满1M，所以返回的消息总字节数是小于等于这个设定值。

因此，这个maxSize至少要比你的最大消息还要大上20个字节（消息的额外头部大小），例如你最大的消息大小可能是1024，那么这个maxSize不能小于1044。通常我会建议将这个参数设置成最大消息大小的2-3倍。这样可以做批量拉取，并且不会导致重复消费的消息太多。

subscribe方法可订阅多个topic，但是对同一个topic只能调用一次：

consumer.subscribe("topic1",1024,messageListener1);
    consumer.subscribe("topic2",1024,messageListener1);
    consumer.subscribe("topic3",1024,messageListener2);

上面的代码同时订阅了topic1,topic2,topic3，其中1和2采用同一个消息处理器，而3采用自己的消费处理器。

在调用了subsribe订阅消息后，订阅过程并没有真正开始，==还需要调用completeSubscribe来使订阅生效。==subscribe支持链式调用：

consumer.subscribe("topic1",1024,messageListener1).subscribe("topic2",1024,messageListener1).completeSubscribe();

当completeSubscribe正常完成的情况下，消费者将从zookeeper上查找到提供这些topic的服务器并建立TCP链接，然后启动抓取线程，按照maxSize的大小抓取消息，抓取后的消息解析出来后提交给消息处理器的receiveMessages方法处理。

## 回滚消息 ##

MetaQ的Consumer会在两种情况下回滚一条消息并重试消费：

*  recieveMessages抛出任何运行时异常，都将导致消费回滚并重试。
 *  在消息处理过程中调用Messgae.setRollbackOnly()方法主动回滚消息。  
    消息被回滚后将会尝试重新调用recieveMessages做消费，如果连续消费失败超过maxFetchRetries设定的阈值，将会调用拒绝处理器RejectConsumptionHandler来处理。

## 消息过滤 ##

引入了新的`ConsumerMessageFilter`接口，用于消费者过滤消息，在某些场景下你可能只想消费一个topic下满足一定要求的消息：

MessageConsumer subscribe(String var1, int var2, MessageListener var3,ConsumerMessageFilter var4) throws MetaClientException;
    
    public interface ConsumerMessageFilter { 
        boolean accept(String var1, Message var2);
    }

这个接口的实现必须是线程安全的，抛出任何异常都将被认为消息不被接受，也就是忽略消费。如果可以消费，返回true就可。

比如下面这个实现，只接收消息属性为accept的消息：

package com.taobao.metamorphosis.example.filter;
    
    import com.taobao.metamorphosis.Message;
    import com.taobao.metamorphosis.consumer.ConsumerMessageFilter;
    
    
    public class AttributeMessageFilter implements ConsumerMessageFilter { 
    
        @Override
        public boolean accept(String group, Message message) { 
            if (message.getAttribute() == null) { 
                return false;
            }
            return message.getAttribute().equals("accept");
        }
    
    }

在订阅的subscribe方法传入该实现：

messageConsumer.subscribe(topic, 1024*1024, messageListener, new AttributeMessageFilter());

# 通信协议 #

Meta是通讯是走TCP长连接，它的的协议是基于文本行的协议，类似memcached的文本协议。通用的协议格式如下

command params opaque\r\n
    [body]

其中command为协议命令，params为参数列表，而opaque为协议的自增序列号，用于请求和应答的映射。客户端发送协议的时候需要自增此序列号，而服务端将拷贝来自客户端的序列号并作为应答的序列号返回，客户端可根据应答的序列号将应答和请求对应起来。body为协议体，可选，在协议头里需要必须有字段指明body长度。

## TCP长连接 ##

1.  短连接的操作步骤是：  
    建立连接——数据传输——关闭连接…建立连接——数据传输——关闭连接
2.  长连接的操作步骤是：  
    建立连接——数据传输…（保持连接）…数据传输——关闭连接

## 长短连接的优点和缺点 ##

*  长连接可以省去较多的TCP建立和关闭的操作，减少浪费，节约时间。
 *  对于频繁请求资源的客户来说，较适用长连接。
 *  不过这里存在一个问题，存活功能的探测周期太长，还有就是它只是探测TCP连接的存活，属于比较斯文的做法，遇到恶意的连接时，保活功能就不够使了。在长连接的应用场景下，client端一般不会主动关闭它们之间的连接，
 *  client与server之间的连接如果一直不关闭的话，会存在一个问题，随着客户端连接越来越多，server早晚有扛不住的时候，这时候server端需要采取一些策略，如关闭一些长时间没有读写事件发生的连接，这样可以避免一些恶意连接导致server端服务受损；
 *  如果条件再允许就可以以客户端机器为颗粒度，限制每个客户端的最大长连接数，这样可以完全避免某个蛋疼的客户端连累后端服务。
 *  短连接对于服务器来说管理较为简单，存在的连接都是有用的连接，不需要额外的控制手段。
 *  但如果客户请求频繁，将在TCP的建立和关闭操作上浪费时间和带宽。

# 配置管理 #

从1.4.5开始，MetaQ提供了一个Web管理平台，默认运行在8120端口，你可以通过浏览器访问http://localhost:8120 来访问web管理平台，localhost为本机IP，可替换为broker运行机器所在ip或者hostname。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 3]  
zookeeper配置管理：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 4]  
配置管理可以参考链接：[https://github.com/killme2008/Metamorphosis/wiki/配置管理][https_github.com_killme2008_Metamorphosis_wiki]

*  brokerId: 服务器集群中唯一的id，必须为整型0-1024之间。对服务器集群的定义是使用同一个zookeeper并且在zookeeper上的root path相同，具体参见zookeeper配置。
 *  unflushThreshold: 每隔多少条消息做一次磁盘sync，强制将更改的数据刷入磁盘。默认为1000。也就是说在掉电情况下，最多允许丢失1000条消息。可设置为0，强制每次写入都sync。在设置为0的情况下，服务器会自动启用group commit技术，将多个消息合并成一次sync来提升IO性能。经过测试，group commit情况下消息发送者的TPS没有受到太大影响，但是服务端的负载会上升很多。
 *  unflushInterval: 间隔多少毫秒定期做一次磁盘sync,默认是10秒。也就是说在服务器掉电情况下，最多丢失10秒内发送过来的消息。不可设置为小于或者等

# MetaQ最佳实践 #

## 生产者最佳实践 ##

*  尽量服用MessageProducer。可以单个MessageProducer发送多种topic，或者多个MessageProducer每个发送一种topic，前提是不要重复创建。
 *  消息data的序列化方式建议不要使用特定于语言的序列化方式（如java序列化），可考虑自定义协议、json、protobufs、hessian都有序列化协议，以便跨语言消费。
 *  实现发送顺序所需要的ParitionSelector，我们会推荐使用业务id，如交易订单id来取模分区列表选择固定分区发送。
 *  单条消息大小最好限制在100k以下。
 *  如无顺序等特殊要求，不要实现自己的PartitionSelector,默认的轮询策略足够。

## 消费者最佳实践 ##

*  尽量复用MessageConsumer，可以单个MessageConsumer订阅多种topic，或者多个MessageConsumer每个订阅一种topic，前提是不要重复创建。
 *  单次拉取的数据不宜过大，对消息实时性要求较高的应用，应将单次拉取的数据缩小，但至少大于单条消息的大小。如对吞吐量要求较高，可将该值设大。
 *  如消费过程非常轻量级(比如只是打印)，可不设置MessageListener线程池，减少资源耗费
 *  如消息发送量巨大，消费能力不高，可适当提高拉取消息线程数fetchRunnerCount和MessageListener的线程池大小。
 *  尽量在消息消费过程中捕捉所有异常，减少消息在本地的堆积和恢复，前提是不要遗漏消息。如确实无法处理，请主动抛出运行时异常以便重试。

[20190702121629151.png]: /images/20211219/e7dac4adcc5f44ff80a262e207bea87f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70]: /images/20211219/bc8938aae5604dbfa6cd839eba6aa266.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 1]: /images/20211219/e43b3bc638684bffb41d4ab6f1184de7.png
[20190702121802610.png]: /images/20211219/c1c0c7ed7af5447eab8c549a376ea897.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 2]: /images/20211219/313ee5cff5f248879eea3e8b13b4c752.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 3]: /images/20211219/f997059e7646474aa4d1296502326bb3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNjM3Mjg3OQ_size_16_color_FFFFFF_t_70 4]: /images/20211219/61afadb9a0954e97b6f0c57969d50287.png
[https_github.com_killme2008_Metamorphosis_wiki]: https://github.com/killme2008/Metamorphosis/wiki/%E9%85%8D%E7%BD%AE%E7%AE%A1%E7%90%86