共计 8814 个字符，预计需要花费 23 分钟才能阅读完成。
这篇文章给大家分享的是有关 RocketMQ 如何安装使用的内容。丸趣 TV 小编觉得挺实用的，因此分享给大家做个参考，一起跟随丸趣 TV 小编过来看看吧。
一、服务端安装部署
我是在虚拟机中的 CentOS6.5 中进行部署。
1. 下载程序
2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解压到适当的目录如 /opt/ 目录
3. 启动 RocketMQ：进入 rocketmq/bin 目录 执行
nohup sh mqnamesrv 4. 启动 Broker，设置对应的 NameServer
nohup sh mqbroker -n 127.0.0.1:9876 二、编写客户端
可以查看 sameple 中的 quickstart 源码 1.Consumer 消息消费者
/**
* Consumer，订阅消息
*/
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer( QuickStartConsumer
consumer.setNamesrvAddr( 127.0.0.1:9876
consumer.setInstanceName( QuickStartConsumer
consumer.subscribe( QuickStart , *
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List MessageExt msgs,
ConsumeConcurrentlyContext context) { System.out.println(Thread.currentThread().getName() + Receive New Messages: + msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println( Consumer Started.
}
}2.Producer 消息生产者
/**
* Producer，发送消息
*
*/
public class Producer { public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer( QuickStartProducer
producer.setNamesrvAddr( 127.0.0.1:9876
producer.setInstanceName( QuickStartProducer
producer.start();
for (int i = 0; i 1000; i++) {
try {
Message msg = new Message( QuickStart ,// topic
TagA ,// tag
(Hello RocketMQ ,QuickStart + i).getBytes()// body
);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
catch (Exception e) { e.printStackTrace();
Thread.sleep(1000);
}
}
producer.shutdown();
}
}3. 首先运行 Consumer 程序，一直在运行状态接收服务器端推送过来的消息
23:18:07.587 [main] DEBUG i.n.c.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16
23:18:07.591 [main] DEBUG i.n.util.internal.PlatformDependent - Platform: Windows
23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - Java version: 7
23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noUnsafe: false
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.ByteBuffer.cleaner: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Buffer.address: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available
23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: true
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - sun.misc.Unsafe: available
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noJavassist: false
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - Javassist: unavailable
23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent -
You don t have Javassist in your class path or you don t have enough permission to load dynamically generated classes.
Please check the configuration for better performance.
23:18:07.595 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false
23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false
23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512
23:18:08.355 [main] DEBUG i.n.util.internal.ThreadLocalRandom - -Dio.netty.initialSeedUniquifier: 0x8c0d4793e5820c31
23:18:08.446 [NettyClientWorkerThread_1] DEBUG io.netty.util.ResourceLeakDetector
- -Dio.netty.noResourceLeakDetection: false
Consumer Started.4. 再次运行 Producer 程序，生成消息并发送到 Broker，Producer 的日志冲没了，但是可以看到 Broker 推送到 Consumer 的一条消息
ConsumeMessageThread-QuickStartConsumer-3 Receive New Messages: [MessageExt [queueId=0, storeSize=150,
queueOffset=244, sysFlag=0, bornTimestamp=1400772029972, bornHost=/10.162.0.7:54234,
storeTimestamp=1400772016017, storeHost=/127.0.0.1:10911, msgId=0A0A0A5900002A9F0000000000063257,
commitLogOffset=406103, bodyCRC=112549959, reconsumeTimes=0, preparedTransactionOffset=0,
toString()=Message [topic=QuickStart, flag=0,
properties={TAGS=TagA, WAIT=true, MAX_OFFSET=245, MIN_OFFSET=0}, body=29]]]三、Consumer 最佳实践
1. 消费过程要做到幂等（即消费端去重）
RocketMQ 无法做到消息重复，所以如果业务对消息重复非常敏感，务必要在业务层面去重，有以下一些方式：
（1）. 将消息的唯一键，可以是 MsgId，也可以是消息内容中的唯一标识字段，例如订单 ID，消费之前判断是否在 DB 或 Tair（全局 KV 存储）中存在，如果不存在则插入，并消费，否则跳过。（实践过程要考虑原子性问题，判断是否存在可以尝试插入，如果报主键冲突，则插入失败，直接跳过）msgid 一定是全局唯一的标识符，但是可能会存在同样的消息有两个不同的 msgid 的情况（有多种原因），这种情况可能会使业务上重复，建议最好使用消息体中的唯一标识字段去重
（2）. 使业务层面的状态机去重
2. 批量方式消费
如果业务流程支持批量方式消费，则可以很大程度上的提高吞吐量，可以通过设置 Consumer 的 consumerMessageBatchMaxSize 参数，默认是 1，即一次消费一条参数
3. 跳过非重要的消息
发生消息堆积时，如果消费速度一直跟不上发送速度，可以选择丢弃不重要的消息
@Override
public ConsumeConcurrentlyStatus consumeMessage(List MessageExt msgs,
ConsumeConcurrentlyContext context) { System.out.println(Thread.currentThread().getName() + Receive New Messages: + msgs);
long offset=msgs.get(0).getQueueOffset();
String maxOffset=msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
long diff=Long.parseLong(maxOffset)-offset;
if(diff 100000){
// 处理消息堆积情况
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}如以上代码所示，当某个队列的消息数堆积到 100000 条以上，则尝试丢弃部分或全部消息，这样就可以快速追上发送消息的速度
4. 优化没条消息消费过程
举例如下，某条消息的消费过程如下
1. 根据消息从 DB 查询数据 1
2. 根据消息从 DB 查询数据 2
3. 复杂的业务计算
4. 向 DB 插入数据 3
5. 向 DB 插入数据 4
这条消息的消费过程与 DB 交互了 4 次，如果按照每次 5ms 计算，那么总共耗时 20ms，假设业务计算耗时 5ms，那么总过耗时 25ms，如果能把 4 次 DB 交互优化为 2 次，那么总耗时就可以优化到 15ms，也就是说总体性能提高了 40%。
对于 Mysql 等 DB，如果部署在磁盘，那么与 DB 进行交互，如果数据没有命中 cache，每次交互的 RT 会直线上升，如果采用 SSD，则 RT 上升趋势要明显好于磁盘。
个别应用可能会遇到这种情况：在线下压测消费过程中，db 表现非常好，每次 RT 都很短，但是上线运行一段时间，RT 就会变长，消费吞吐量直线下降
主要原因是线下压测时间过短，线上运行一段时间后，cache 命中率下降，那么 RT 就会增加。建议在线下压测时，要测试足够长时间，尽可能模拟线上环境，压测过程中，数据的分布也很重要，数据不同，可能 cache 的命中率也会完全不同
四、Producer 最佳实践
1. 发送消息注意事项
（1）一个应用尽可能用一个 Topic，消息子类型用 tags 来标识，tags 可以由应用自由设置。只有发送消息设置了 tags，消费方在订阅消息时，才可以利用 tags 在 broker 做消息过滤。
（2）每个消息在业务层面的唯一标识码，要设置到 keys 字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过 topic，key 来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证 key 尽可能唯一，这样可以避免潜在的哈希冲突。
（3）消息发送成功或者失败，要打印消息日志，务必要打印 sendresult 和 key 字段
（4）send 消息方法，只要不抛异常，就代表发送成功。但是发送成功会有多个状态，在 sendResult 里定义
SEND_OK：消息发送成功
FLUSH_DISK_TIMEOUT：消息发送成功，但是服务器刷盘超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
FLUSH_SLAVE_TIMEOUT：消息发送成功，但是服务器同步到 Slave 时超时，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失
SLAVE_NOT_AVAILABLE：消息发送成功，但是此时 slave 不可用，消息已经进入服务器队列，只有此时服务器宕机，消息才会丢失。对于精确发送顺序消息的应用，由于顺序消息的局限性，可能会涉及到主备自动切换问题，所以如果 sendresult 中的 status 字段不等于 SEND_OK，就应该尝试重试。对于其他应用，则没有必要这样
（5）对于消息不可丢失应用，务必要有消息重发机制
2. 消息发送失败处理
Producer 的 send 方法本身支持内部重试，重试逻辑如下：
（1）至多重试 3 次
（2）如果发送失败，则轮转到下一个 Broker
（3）这个方法的总耗时时间不超过 sendMsgTimeout 设置的值，默认 10s 所以，如果本身向 broker 发送消息产生超时异常，就不会再做重试
如果调用 send 同步方法发送失败，则尝试将消息存储到 db，由后台线程定时重试，保证消息一定到达 Broker。
上述 db 重试方式为什么没有集成到 MQ 客户端内部做，而是要求应用自己去完成，基于以下几点考虑：
（1）MQ 的客户端设计为无状态模式，方便任意的水平扩展，且对机器资源的消耗仅仅是 cpu、内存、网络
（2）如果 MQ 客户端内部集成一个 KV 存储模块，那么数据只有同步落盘才能较可靠，而同步落盘本身性能开销较大，所以通常会采用异步落盘，又由于应用关闭过程不受 MQ 运维人员控制，可能经常会发生 kill -9 这样暴力方式关闭，造成数据没有及时落盘而丢失
（3）Producer 所在机器的可靠性较低，一般为虚拟机，不适合存储重要数据。综上，建议重试过程交由应用来控制。
3. 选择 oneway 形式发送
一个 RPC 调用，通常是这样一个过程
（1）客户端发送请求到服务器
（2）服务器处理该请求
（3）服务器向客户端返回应答
所以一个 RPC 的耗时时间是上述三个步骤的总和，而某些场景要求耗时非常短，但是对可靠性要求并不高，例如日志收集类应用，此类应用可以采用 oneway 形式调用，oneway 形式只发送请求不等待应答，而发送请求在客户端实现层面仅仅是一个 os 系统调用的开销，即将数据写入客户端的 socket 缓冲区，此过程耗时通常在微秒级。
RocketMQ 不止可以直接推送消息，在消费端注册监听器进行监听，还可以由消费端决定自己去拉取数据
/**
* PullConsumer，订阅消息
*/
public class PullConsumer {
//Java 缓存
private static final Map MessageQueue, Long offseTable = new HashMap MessageQueue, Long
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer( PullConsumerGroup
consumer.setNamesrvAddr( 127.0.0.1:9876
consumer.start();
// 拉取订阅主题的队列，默认队列大小是 4
Set MessageQueue mqs = consumer.fetchSubscribeMessageQueues( TopicTestMapBody
for (MessageQueue mq : mqs) { System.out.println( Consume from the queue: + mq);
SINGLE_MQ:while(true){
try {
PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
List MessageExt list=pullResult.getMsgFoundList();
if(list!=null list.size() 100){ for(MessageExt msg:list){ System.out.println(SerializableInterface.deserialize(msg.getBody()));
}
}
System.out.println(pullResult.getNextBeginOffset());
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
switch (pullResult.getPullStatus()) {
case FOUND:
// TODO
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
}
catch (Exception e) { e.printStackTrace();
}
}
}
consumer.shutdown();
}
private static void putMessageQueueOffset(MessageQueue mq, long offset) { offseTable.put(mq, offset);
}
private static long getMessageQueueOffset(MessageQueue mq) { Long offset = offseTable.get(mq);
if (offset != null){ System.out.println(offset);
return offset;
}
return 0;
}刚开始的没有细看 PullResult 对象，以为拉取到的结果没有 MessageExt 对象还跑到群里面问别人，犯 2 了
特别要注意 静态变量 offsetTable 的作用，拉取的是按照从 offset（理解为下标）位置开始拉取，拉取 N 条，offsetTable 记录下次拉取的 offset 位置。
感谢各位的阅读！关于“RocketMQ 如何安装使用”这篇文章就分享到这里了，希望以上内容可以对大家有一定的帮助，让大家可以学到更多知识，如果觉得文章不错，可以把它分享出去让更多的人看到吧！
