MQ整理心得

1、com.alibaba.rocketmq.client.producer.DefaultMQProducer

2、org.apache.rocketmq.client.producer.TransactionMQProducer

Topic

Properties

Body

Keys

Tags

DelayTimeLevel

UserProperty

普通消息

分区有序消息

全局有序消息

延迟消息

事务消息

重试机制

保证机制

  数据同步方式保证
（Master Broker ->Slave Broker）

RocketMQ中的默认生成者实现，用于生产普通消息、顺序消息、单向消息、批量消息、延迟消息

核心属性

生产者启动流程

业务层

消息处理层

通讯层

业务层

DefaultMQProducer

DefaultMQProducerImpl

MQClientAPIImpl

RemotingClient

//发送普通消息
public static void main(String[] args) throws Exception {
//第一步 设置生产者
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
//第二步 初始化MQ地址
producer.setNamesrvAddr("Addresses");
producer.setRetryTimesWhenSendFailed(2);
producer.start();
//第三步 发送消息
Message msg = new Message("Topic", "TagA", "searchKeys", "body".getBytes());
SendResult sendResult = producer.send(msg);
System.out.println(sendResult.toString());
//第四步 关闭MQ
producer.shutdown();
}

Message msg = new Message("Topic", "TagA", "searchKeys", "body".getBytes());
msg.setDelayTimeLevel(2);

不同延迟等级表

    //发送顺序消息
public static void main2(String[] args) throws Exception {
//第一步 设置生产者
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
//第二步 初始化MQ地址
producer.setNamesrvAddr("Addresses");
producer.setRetryTimesWhenSendFailed(2);
producer.start();
int hashKey = 123;
//第三步 发送消息
Message msg = new Message("Topic", "TagA", "searchKeys", "body".getBytes());
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
@Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object hashkeyArg) {
int id = (int) hashkeyArg;
int index = id % mqs.size();
return mqs.get(index);
}
}, hashKey);
System.out.println(sendResult.toString());
//第四步 关闭MQ
producer.shutdown();
}

public static void main(String[] args) throws Exception {
//第一步 初始化生产者
TransactionMQProducer producer = new TransactionMQProducer("ProducerGroupName");
//第二步 初始化MQ地址，配置生产者的各个参数和Broker回调，检查本地事务处理并启动生产者
producer.setNamesrvAddr("Addresses");
producer.setRetryTimesWhenSendFailed(2);

producer.setCheckThreadPoolMinSize(2);
producer.setCheckThreadPoolMaxSize(2);
producer.setCheckRequestHoldMax(200);

producer.setTransactionCheckListener(new TransactionCheckListener() {
private AtomicInteger tramsactionIndex = new AtomicInteger(0);

@Override
public LocalTransactionState checkLocalTransactionState(MessageExt msg) {
System.out.println(msg);

int value = tramsactionIndex.getAndIncrement();
if ((value % 6) == 0) {
throw new RuntimeException("Could not find db");
} else if ((value % 5) == 0) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else if ((value % 4) == 0) {
return LocalTransactionState.COMMIT_MESSAGE;
}
return LocalTransactionState.UNKNOW;
}
});
producer.start();

//第三步：设置本地事务处理器，发送消息
Message msg = new Message("Topic", "TagA", "searchKeys", "body".getBytes());
SendResult sendResult = producer.sendMessageInTransaction(msg, new LocalTransactionExecuter() {
private AtomicInteger tramsactionIndex = new AtomicInteger(0);

@Override
public LocalTransactionState executeLocalTransactionBranch(Message msg, Object arg) {
int value = tramsactionIndex.getAndIncrement();
if ((value % 6) == 0) {
throw new RuntimeException("Could not find db");
} else if ((value % 5) == 0) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else if ((value % 4) == 0) {
return LocalTransactionState.COMMIT_MESSAGE;
}
return LocalTransactionState.UNKNOW;
}
}, null);
System.out.println(sendResult.toString());
}

public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("Addresses");
producer.setInstanceName("instance name");
producer.start();

Message msg = new Message("Topic", "TagA", "searchKeys", "body".getBytes());
producer.sendOneway(msg);
}

 public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("Addresses");
producer.setInstanceName("instance name");
producer.start();
List<Message> messages = Lists.newArrayList();
messages.add(new Message("Topic", "TagA", "searchKeys", "body1".getBytes()));
messages.add(new Message("Topic", "TagA", "searchKeys", "body2".getBytes()));
messages.add(new Message("Topic", "TagA", "searchKeys", "body3".getBytes()));
producer.send(messages);
producer.shutdown();
}

1、正常消费

2、重试消费

3、死信

namesrvAddr: 表示RocketMQ集群Namesrv地址，如果有多个，用分号分开。如：127.0.0.1:9876;127.0.0.2:9876;

clientIP: 使用客户端的程序所在机器的地址

instanceName: 实例名，每个进程的实例名必须不同，否则程序会启动失败。

vipChannelEnable: 表示是否开启VIP通道，VIP通道区别为单独一个端口进行通讯。

clientCallBackExecutorThreads: 客户端回调线程数。等于Netty通讯层回调线程的个数，默认值：Runtime.getRuntime().availableProcessors()，表示当前有效CPU个数

pollNameServerInterval: 获取Topic路由信息间隔，单位ms,默认30000ms

heartbeatBrokerInterval: 客户端和Broker心跳间隔，单位ms,默认30000ms

persistConsumerOffsetInterval: 持久化消费位点时间间隔，单位ms，默认5000ms

defaultMQPullConsumerImpl: 默认Pull消费者的具体实现

consumerGroup: 消费者组名字。

brokerSuspendMaxTimeMillis: 在长轮询模式下，Broker的最大挂起请求时间，不建议修改

consumerTimeoutMillisWhenSuspend: 在长轮询模式下，消费者的最大请求超时时间，必须比brokerSuspendMaxTimeMillis大，不建议修改

consumerPullTimeoutMillis: 消费者Pull消息时Socket的超时时间

messageModel： 消费模式，集群/广播

messageQueueListener: 消息路由信息变化时回调处理监听器，一般在重新平衡是会被调用。

offsetStore: 位点存储模块。集群模式会持久化到Broker中，广播模式会持久在本地文件中。有两个实现类，RemoteBrokerOffsetStore 和LocalFileOffsetStore.

allocateMessageQueueStrategy: 消费Queue分配策略管理器

maxReconsumeTimes: 最大重试次数，可以配置

registerMassageQueueListener(): 注册队列变化监听器，当队列发生变化时会被监听到。

pull(): 从Broker中Pull消息，如果有PullCallBacke参数，表示异步拉取。

pullBlockIfNotFound(): 长轮询方式拉取，如果没有拉取到消息，那么Broker会将请求Hold住一段时间。

updateConsumeOffset(final MessageQueue mq,final long offset): 更新某一个Queue的消费位点。

fetchConsumeOffset(final MessageQueue mq,final boolean fromStore): 查找某个Queue的消费位点。

sendMessageBack(MessageExt msg ,int delayLevel,String brokerName ,String consumerGroup): 如果消费发送失败，则可以将消息重新发回给Broker,这个消费组延迟一段时间可以再消费（也就是重新消费）

fetchSubscribeMessageQueues(final String topic): 获取一个Topic的全部Queue信息。

大部分与DefaultMQPullConsumer是一样的 

defaultMQPushConsumerImpl: 默认的Push消费者具体实现类

consumerFromWhere: 一个枚举值，表示从什么位点开始消费

consumeTimestamp: 表示从哪一时刻开始消费，格式为yyyyMMDDHHmmss,默认为半小时前，当consumerFromWhere=CONSUME_FROM_TIMESTAMP 才生效

allocateMessageQueueStrategy: 消费者订阅topic-queue策略

subscription: 订阅关系，表示当前消费者订阅了哪些Topic的哪些Tag。

messageListener: 消息Push回调监听器

consumeThreadMin: 最小消费线程数，必须小于ConsumeThreadMax。

ConsumeThreadMax: 最大消费线程数，必须大于consumeThreadMin。

adjustThreadPollNumsThreshold: 动态调整消费线程池的线程数大小。（开源版本不支持）

consumeConcurrentlyMaxSpan: 并发消息的最大位点查。如果Pull消息的位点差超过该值，拉取变慢

pullThresholdForQueue: 一个Queue能缓存的最大消息数。超过该值则采取拉取流控措施。

pullThresholdForTopic: 一个Topic最大能缓存的消息字节数，超过则采取流控

pullThresholdSizeForTopic: 一个Topic最大能缓存的消息字节数，单位是MB。默认-1 ，结合pullThresholdForQueue配置项生效

pullInterval: 拉取间隔，单位ms

consumMessageBatchMaxSize: 消费者每次批量消费时，最多消费多少条消息。

pullBatchSize: 一次最多拉取多少条消息

postSubscriptionWhenPull: 每次拉取消息时是否更新订阅关系，默认为false

maxReconsumeTimes:最大重试次数，默认返回-1，表示默认最大为16次

suspendCurrentQueueTimeMillis: 为短轮询场景设置的挂起时间，比如顺序消息的场景

consumeTimeout

1、DefaultMQPullConsumer

2、DefaultMQPullConsumerImpl

3、MQClientInstance

1、DefaultMQPushConsumer

2、DefaultMQPushConsumerImpl

3、MQClientInstance