## 简介
​		苍穹提供对消息中间件MQ的支持（目前支持rabbitmq），消息中间件是企业IT管理系统内部通讯的重要核心系统之一，具有系统解耦，异步化，削峰填谷等作用。



## 特性

- 发布/订阅者模式
  - 1->1，单个订阅者
  - 1->N，广播，所有订阅者
- 可靠性
  - 消息可靠持久化
  - 消息确认机制
  - 消息重新发送
- 消息发送事务支持
- 消息隔离机制
  - vhost 按集群隔离，一般可配置为集群名
  - 消息按照region-queue进行隔离
  - 区域region：可以用appid或领域名
    - 队列queue：按功能
- Region可以独立部署消息服务器
  - mq.server=xxx.xxx.xxx.xxx  公共的
  - mq.server.fi=xxx.xxx.xxx.xxx //region fi独立部署



## 使用指南

### 1. 在配置服务里配置队列及消费者类
- 队列消息默认会持久化到MQ服务上(duration)
- AppId指定该队列消费者在哪个应用上启动，如果没配置或者没分应用部署，将在所有节点启动；业务队列需要配置。
- Consumer指定该队列消息的消费者类


```java
<root>
    <region name="demo" appid="appid">
        <queue name="demo_queue"  appid="appid">
            <consumer class="kd.bos.mq.demo.DemoConsumer" ></consumer>
        </queue>
    </region>
</root>
```

#### 标准产品配置

- 将需要增加修改的配置，发给平台部统一提交管理。

#### 客户开发配置

##### 约束

- 客户开发能增加消费队列，但不能替换标准产品消费者（相同region下如果queue name和标准产品相同，则该queue配置将不会被添加）
- 配置文件需以.xml结尾

##### 配置
  1.	增加xml配置文件，如取名为custommqconfig.xml，放在包erkai下，工程中放以下目录
```java
src/main/resources/erkai/custommqconfig.xml
```
  2.	在配置文件中配置二次开发增加的队列相关信息，如
```xml
<root>
	<region name="fi" >
		<queue name="fi.gl.ercikaifa_queue" appid="gl">
			<consumer class="kd.fi.gl.opplugin.ErKaiConsumer"></consumer>
		</queue>
		
		<queue name="fi.gl.ercikaifa_queue1" appid="gl">
			<consumer class="kd.fi.gl.opplugin.ErKaiConsumer1"></consumer>
		</queue>
	</region>
</root>
```
其中不同region的配置需要放在不同的配置文件中.

  3.	打包部署jar包后，同时将配置到MC中添加配置项：key为mqConfigFiles.config，值为erkai.custommqconfig.xml，如果有多个文件，换行。


### 2. 接口描述

#### 消息消费者实现接口 
- kd.bos.mq.MessageConsumer
```java
public interface MessageConsumer {
/**
 * @param message 消息内容
 * @param messageId 消息id
 * @param acker 消息确认器
 * @param resend 是否重新发送的，即前面发送的没有确认成功，mq服务器会重复发送过来
 */
public void onMessage(Object message, String messageId, boolean resend, MessageAcker acker);
}
```

#### 消息确认接口MessageAcker

```java
public interface MessageAcker {
	/**
	 * 确认处理完毕，调用ack从队列中删除该消息
	 * @param messageId
	 */
	void ack(String messageId);
	/**
	 * 拒绝，返回队列，队列将再次发生给消息消费者
	 * @param messageId
	 */
	void deny(String messageId);
	
	/**
	 * 丢弃消息，队列将丢弃该条消息
	 * @param messageId
	 */
	void discard(String messageId);
}
```

#### 消息发送接口

- MessagePublisher

```java
public interface MessagePublisher {
	
	public void publish(byte[] message);
	public void publish(String message);
	public void publish(Object message);
	

	/**
	 * publish message in db transcation 
	 * @param routKey 该参数没用了
	 * @param message
	 */
	public void publishInDbTranscation(String routKey,Object message);
	
	/**
	 * 用完需要关闭
	 */
	public void close();
}
```

### 3. 示例代码
#### 消息消费者
- 实现接口 kd.bos.mq.MessageConsumer
```java
public class DemoConsumer implements MessageConsumer{
    public void onMessage(Object body, String messageId, boolean resend, 		MessageAcker acker) {
        try {
            String str = (String)body; //new String(bytes, "UTF-8");
            logger.info(str);
            RequestContext rc = RequestContext.get(); //获取上下文
            acker.ack(messageId);
        } catch (Throwable e) {
            boolean discard = e instanceof IOException; //是否废弃这条消息，根据具体场景判断
            if (discard){
                acker.discard(messageId);//废弃
                // 记录废弃原因，并写业务日志
            } else{
                acker.deny(messageId);//告诉mq重发这条消息
                // 记录异常原因，并写业务日志
            }
        }
    }
    //当事务敏感，发送方和消费方不在同一个库时，需要重写设置消费方的routekey
    String getRouteKey()
    {
        return  "consumerDbRouteKey”;
    }
}
```

##### 消息确认MessageAcker

​		在具体消息消费者类的onMessage方法中，如果没有调用MessageAcker的方法，默认行为为：当onMessage正常执行，该消息正常消费（acker.ack(messageId)）；当onMessage方法抛出异常，则记录日志并丢弃该消息（acker.discard(messageId)）。

​        在业务消费者类onMessage方法的实现中，如示例代码所示，需要捕获所有异常，判断具体异常原因，可恢复异常调用 acker.deny(messageId)告诉mq重发这条消息，不可恢复异常调用acker.discard(messageId)丢弃该消息。记录异常原因，并写入业务日志记录。

#### 消息发送示例
- 区分事务和非事务接口，事务接口需在事务中调用。
- 目前没有广播接口
```java
支持事务（当前代码在事务环境中）：
Trascation begin
MessagePublisher pub = MQFactory.get().createSimplePublisher("demo", "demo_queue");
		try
		{
            //第一个参数现在不用了，不需要传
			pub.publishInDbTranscation(null, "hello world");
		} finally
		{
            //用完关闭
			pub.close();
		}
Transcation end
```

```java
不支持事务：
MessagePublisher pub = MQFactory.get().createSimplePublisher("demo", "demo_queue");
try
{
	pub.publish(“hello world”);
} finally{
    //用完关闭
	pub.close();
}

```

### 4. 高级特性

- Tips
  - 在Consumer中获取RequestContext，该RequestContext对应了发送者的RequestContext，可以在其中获取租户id，用户id等等
  - 如果消息发送方没有RequestContext，消息将不能被消费，所以如果自建线程中发送mq消息，需要传递或创建RequestContext。
  - 在消费者代码中，消费者消费完后，应该调用ack方法，如果消费异常，需要进行判断当前消息需要重新消费，还是需要被丢弃
  - 任何情况下，只要消息没有确认成功，如网络异常等，MQ服务器会重新发送消息，resend为true所以，消费者必须处理幂等情况，及支持重复消费消息
### 5.调试及常见问题

- 调试

  - 默认情况下，消息会发送给任意一个消费者节点，在开发调试阶段无法保证自己发送的message被自己java进程消费到
  - 在DebugServer中加一个参数：
```java
	System.setProperty("mq.debug.queue.tag", "lzx");
```
​		该参数必须全局唯一，才能确保被自己的java进程接收到。

​		有时为分析问题需要连接部署环境进行调试，为了不影响部署环境的数据，需要添加参数：

```java
	System.setProperty("dubbo.registry.register", "false");
	System.setProperty(" mq.consumer.register”, “false");
```
​		如果就是要调试mq消费的代码块，需要设置
```java
	System.setProperty("dubbo.registry.register", "true");
	System.setProperty("mq.debug.queue.tag", "随机字符串");
```

- 常见问题

  - 现场开发环境收不到mq消息,检查下面两项配置

    ```java
    System.setProperty("mq.consumer.register", "true");//该参数为false，本节点将不会消费mq消息
    System.setProperty("mq.debug.queue.tag", "每个人的专属tag");
    ```

  - 现场测试环境或其它线上环境建发性收不到mq消息：

    - 开发环境为调试问题连接到了测试环境，而又没合适的配置开发环境，导致测试环境的mq消息被开发人员的环境消费，此时开发环境需要设置(如开发人员太多，有必要禁止开发人员直连测试环境)：

      ```java
    System.setProperty("mq.consumer.register", "true");//该参数为false，本节点将不会消费mq消息
      System.setProperty("mq.debug.queue.tag", "每个人的专属tag");
      ```
  
    - 测试环境和其它环境共用vhost，导致测试环境的消息被其它环境消费，此时需要检查所有环境的vhost设置，保证每个环境使用独立的vhost（一般可用集群名）

      ```java
      在配置中心查询mq.server 检查vhost设置，若实在环境太多检查不过来，将当前环境vhost设置一个新的值，并在rabbitmq管理中心创建该vhost。
      ```



