MQ42# RocketMQ消息轨迹

发表于 2020-12-20 更新于 2021-02-10 分类于 RocketMQ

Broker配置

首先看下broker.conf配置的两个属性

| 属性 |默认值 |

| — | — |

|traceTopicEnable |false |

| msgTraceTopicName | RMQ_SYS_TRACE_TOPIC |

在一个集群中可以配置一台机器专门负责消息轨迹的收集工作，该台机器上配置traceTopicEnable = true,

borker启动的时候自动创建默认轨迹topic

TopicConfigManager.java构造方法，BrokerController在启动的时候，会初始化TopicConfigManager实现trace topic的创建工作

{

if (this.brokerController.getBrokerConfig().isTraceTopicEnable()) {

String topic = this.brokerController.getBrokerConfig().getMsgTraceTopicName();

TopicConfig topicConfig = new TopicConfig(topic);

this.systemTopicList.add(topic);

topicConfig.setReadQueueNums(1);

topicConfig.setWriteQueueNums(1);

this.topicConfigTable.put(topicConfig.getTopicName(), topicConfig);

}

}

阅读全文 »

MQ41# RocketMQ特性--事物消息

发表于 2020-12-20 更新于 2021-02-10 分类于 RocketMQ

问题思考

从官方给的例子入手，代码如下:

示例类：org.apache.rocketmq.example.transaction.TransactionProducer.java

public static void main(String[] args) throws MQClientException, InterruptedException {

//@1 定义TransactionListener

TransactionListener transactionListener = new TransactionListenerImpl();

//@2 使用事务发送Producer

TransactionMQProducer producer = new TransactionMQProducer("please_rename_unique_group_name");

//@3 定义线程池

ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {

@Override

public Thread newThread(Runnable r) {

Thread thread = new Thread(r);

thread.setName("client-transaction-msg-check-thread");

return thread;

}

});

//设置线程池

producer.setExecutorService(executorService);

//设置监听器

producer.setTransactionListener(transactionListener);

producer.setNamesrvAddr("127.0.0.1:9876");

//@4 发送者启动

producer.start();

String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};

for (int i = 0; i < 10; i++) {

try {

Message msg =

new Message("TopicTest1234", tags[i % tags.length], "KEY" + i,

("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

//@5 消息发送

SendResult sendResult = producer.sendMessageInTransaction(msg, null);

System.out.printf("%s%n", sendResult);

Thread.sleep(10);

} catch (MQClientException | UnsupportedEncodingException e) {

e.printStackTrace();

}

}

for (int i = 0; i < 100000; i++) {

Thread.sleep(1000);

}

//发送者关闭

producer.shutdown();

}

阅读全文 »

MQ40# RocketMQ一次消费性能问题排查

发表于 2020-12-20 更新于 2021-02-10 分类于 RocketMQ

需求描述

在容器推广中，为了测试容器的性能，需要消息SDK与ECS上在发送和消费的性能对比；在对比消费性能时，发现容器中的消费性能居然是ECS的2倍。容器并发消费的20个线程TPS在3万左右，ECS中20个消费线程TPS在1.5万左右。

问题：配置均采用8C16G，容器中的性能几乎是ECS的两倍，这不科学，事出反常必有妖。

阅读全文 »

MQ39# RocketMQ消息发送Broker端流程处理

发表于 2020-12-20 更新于 2021-02-10 分类于 RocketMQ

Broker处理消息的入口类SendMessageProcessor

processRequest方法主要三件事情：

1.处理consumer发回broker的消息重试

2.处理批量发送

3.处理单条消息发送

阅读全文 »

MQ38# RocketMQ消息发送（二）

发表于 2020-12-20 更新于 2021-02-10 分类于 RocketMQ

发送接口分类

* 按照发送方式分类

同步发送：等待返回结果
异步发送：异步回调发送结果
一次发送：无结果返回

* 按一次发送消息数量分类

单条消息发送
批量消息发送

* 按照是否指定MessageQueue分类

随机选择发送
指定特定MessageQueue
自定义MessageQueue选择器

阅读全文 »

MQ37# RocketMQ消息发送（一）

发表于 2020-12-20 更新于 2021-02-10 分类于 RocketMQ

消息发送代码

* 需要设置produerGroup

* 需要设置NameServer地址

DefaultMQProducer producer = new DefaultMQProducer("melon-tst");

producer.setNamesrvAddr("localhost:9876");

producer.setVipChannelEnabled(false);

producer.start();

for(int i=0;i<100;i++){

Message msg = new Message("topic_online_test",("Hello RocketMQ" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));

//msg.setDelayTimeLevel(10);

SendResult sendResult = producer.send(msg);

System.out.printf("%s%n",sendResult);

}

producer.shutdown();