RocketMQ实战入门

（一）安装与启动

下载rocketmq-4.8.0二进制包，启动name server和broker：

nohup sh /usr/rocketmq-4.8.0/bin/mqnamesrv > /dev/null 2>&1 & 
nohup sh /usr/rocketmq-4.8.0/bin/mqbroker -n localhost:9876 -c conf/broker.conf > /dev/null 2>&1 &

注：RocketMQ默认需求的jvm内存比较大，可以酌情减少，修改runserver.sh、runbroker.sh两个启动脚本里边关于jvm内存参数的设置。

修改broker.conf配置文件，添加：

namesrvAddr=外网IP:9876
brokerIP1=外网IP

这样是为了能够使客户端能够远程的访问mq

停止mq：

./mqshutdown broker
./mqshutdown namesrv

（二）添加后台管理端

修改/etc/profile，添加环境变量：

export NAMESRV_ADDR=localhost:9876

然后source /etc/profile使配置环境变量生效。

到https://github.com/apache/rocketmq-externals下载rocketmq-console项目源代码，修改配置文件application.properties启用web端登陆密码验证，并在users.properties里边配置ACL的用户和密码：

rocketmq.config.loginRequired=true #开启web端登录密码验证

# Define Admin
admin=password,1

# Define Users
#user1=user1
#user2=user2
user1=query

maven编译打包：

mvn clean package -Dmaven.test.skip=true

启动应用:

nohup java -jar rocketmq-console-ng-2.0.0.jar --server.port=19876 --rocketmq.config.namesrvAddr=127.0.0.1:9876 > /dev/null 2>&1 & 

访问地址： http://122.51.112.187:19876/#/login

（三）ACL访问控制列表

默认搞好的RocketMQ是任意客户端只要网络通的话都可以进行发布订阅的，而ACL机制可以使得mq对客户端的访问进行校验。mq服务端需要配置broker.conf和plain_acl.yml两个配置文件，前者是配置开启acl=true，后者是配置acl具体的账户与权限信息，权限信息支持设置的很细，具体到topic和group级别。

broker.conf

brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
namesrvAddr=访问IP:9876
brokerIP1=访问IP
aclEnable=true

plain_acl.yml

globalWhiteRemoteAddresses:

accounts:
- accessKey: wanganmq
  secretKey: rocket008
  whiteRemoteAddress:
  admin: false
  defaultTopicPerm: PUB|SUB
  defaultGroupPerm: PUB|SUB
  topicPerms:

  groupPerms:
  # the group should convert to retry topic

- accessKey: mqadmin
  secretKey: password
  whiteRemoteAddress:
  # if it is admin, it could access all resources
  admin: true

（四）实践心得

1、推荐开启ACL，然后admin账户交给管理员通过console和cli工具进行管理使用，应用中配置普通账户，这样方便进行管理。

2、一定要开发规范中约定好topic和tag，以及producer/consumer group这些个条目的命名规范，以及约定代表的含义。

例如，topic和tag首字母大写驼峰，分别表示业务域和业务域所发生的事件。用“应用名-producer-group”来命名生产者group名字，"业务处理名-consumer-group" 等等。 见名知意的命名能够提高团队开发效率。

3、跟其他开源中间件一样，了解底层实现原理永远都是对的。不仅出现问题能够搞定和排查，应用实践上也能更给出最佳实践。

4、关于每个Consumer启动多少个线程并发去消费。RocketMQ源码里写死默认是20个线程，用的是jdk线程池。可以调整。DefaultMQPushConsumer.setConsumeThreadMin(4)

附： 搞acl时的一个小插曲，RocketMQ消息发布报错：No accessKey is configured

No accessKey is configured 本来这个错误是很简单个错误，本意是mq上边开启了acl验证机制，而客户端没有配置accessKey。但是笔者这次遇到的这个问题比较有意思，废了我半天时间去研究。在这里记录一下。

当时笔者是先行自己封装好了一个rocket的工具库，考虑到安全问题，决定引入acl机制，然后主要是参照了网上的两个文章。配置mq的broker和plain_acl两个配置文件，客户端加入acl相关代码，主要是加入了AclRPCHook，在发送消息之前插入accessKey和签名供mq进行校验。之后用测试工程引用本地maven库上安装的rocket工具库进行测试，结果之前能用的代码在用了acl之后一直报错：提示accessKey没配置。

然后发现直接弄个带main方法的类运行例子里的代码是可以的，一样的代码copy到测试工程里就不行了。至此确认应该是客户端问题，mq的acl配置应该是对的。
接下来1天各种实验无果，开始回到起点，“源码之下无秘密”，潜心来看源代码，搞清楚客户端的mq发送消息以及acl的源码是怎么写的。

当调用DefaultMQProducer的send方法，同步的向mq投递消息的时候，实际上是defaultMQProducerImpl.send(msg)，也即DefaultMQProducerImpl的sendKernelImpl方法，关键代码：sendResult = this.mQClientFactory.getMQClientAPIImpl().sendMessage，里边其实是MQClientAPIImpl的如下方法：

private SendResult sendMessageSync(
        final String addr,
        final String brokerName,
        final Message msg,
        final long timeoutMillis,
        final RemotingCommand request
    ) throws RemotingException, MQBrokerException, InterruptedException {
        RemotingCommand response = this.remotingClient.invokeSync(addr, request, timeoutMillis);
        assert response != null;
        return this.processSendResponse(brokerName, msg, response,addr);
    }

其中remotingClient是接口，其实现是netty NettyRemotingClient invokeSync()，后者会doBeforeRpcHooks(addr, request);

if (rpcHooks.size() > 0) {
            for (RPCHook rpcHook: rpcHooks) {
                rpcHook.doBeforeRequest(addr, request);
            }
        }

rpcHooks这里相当于是NettyRemotingClient的成员变量。

对于acl的DefaultMQProducer
DefaultMQProducer实例化的时候，是把rpcHook都给到了自己的defaultMQProducerImpl。

factory是在defaultMQProducerImpl.start()也就是DefaultMQProducer.start()的时候实例化的。

MQClientInstance是factory的实现，实例化的时候内部,

this.mQClientAPIImpl = new MQClientAPIImpl(this.nettyClientConfig, this.clientRemotingProcessor, rpcHook, clientConfig);

然后factory提供这个mQClientAPIImpl实例。

defaultMQProducerImpl的start方法中实例化factory时，会判断如果生成过，就不再生成了。也就是说整个进程只会有一个factory实例。关键代码：

this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);

到这里，笔者搞清楚问题了，刚好工程里有另外一个transactionProducer的代码，是笔者用来测试事务消息的，没加acl代码，所以一旦先transactionProducer.start()创建了factory，那之后一直用这个无acl的，所以发送消息到mq的时候一直提示没有accessKey就是这个原因。

（五）Spring整合RocketMQ开发

编写自己的rocketutil开发包，然后maven install到本地maven库上。需要使用的gradle项目在repositories添加mavenLocal()

整体思路：使用集群消息模式，然后每个应用服务使用1个单例的GeneralMqProducer工具类，用于发送消息。再提供一个MQMsgHandler接口，用于应用服务自己编写接口实现：声明订阅消息topic和tag，收到订阅消息之后需要回调执行的业务逻辑。MQMsgHandler接口的实现类、也可以称为是消费者添加自定义注解MsgConsumer、通过Spring BeanPostProcessor机制加载到Spring容器。

消息Pojo：

/**
 *  用于在MQ中进行传输的事件消息
 * */
@Data
public class EventMessage {
    private String msgId; //消息的唯一标识
    private String producerGroup; //消息的生产者所属的分组
    private String topic;   //消息所属的主题（业务域）
    private String tag;     //消息tag，即该主题下的什么事件
    private String msgBody; //消息体, 一般为json string
    private String publishTime; //发布时间
    
}

生产者工具类：

/**
 *  每个应用有一个producer单例。
 *  producergroup表示消息由哪个应用集群投递。
 *  topic代表消息属于哪个业务域的消息，tag代表该业务域下的哪种事件发生了。
 *  msgId是消息的唯一标识
 * */
@Slf4j
@Component
public class GeneralMqProducer {
    
    @Value("${rocketmq.url}")
    private String mqurl;
    
    @Value("${rocketmq.accessKey}")
    private String accessKey;
    
    @Value("${rocketmq.secretKey}")
    private String secretKey;
    
    @Value("${rocketmq.producergroup.name}")
    private String producerGroupName;
    
    private DefaultMQProducer producer;
    
    /**
     *  同步发布消息
     * */
    public SendResult syncPublish(EventMessage eventMsg) {
        try {
            Message msg = new Message(eventMsg.getTopic(), eventMsg.getTag(), eventMsg.getMsgId(), eventMsg.getMsgBody().getBytes("utf-8"));
            
            SendResult sendResult = producer.send(msg);
            return sendResult;
        } catch (Exception e) {
            log.error("向RocketMQ发布消息失败：" + e.getMessage(), e);
            e.printStackTrace();
        }
        return null;
    }
    
    /**
     *  异步发布消息
     *  SendCallback为消息发送完毕后的回调方法
     * */
    public void asyncPublish(EventMessage eventMsg, SendCallback callBack) {
        try {
            eventMsg.setMsgId(UUID.randomUUID().toString().replaceAll("-",""));
            eventMsg.setProducerGroup(producerGroupName);
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss:SSS");
            eventMsg.setPublishTime(LocalDateTime.now().format(formatter));
            
            Message msg = new Message(eventMsg.getTopic(),
                                      eventMsg.getTag(), 
                                      eventMsg.getMsgId(), 
                                      JSON.toJSONString(eventMsg).getBytes("utf-8"));
            
            producer.send(msg, callBack);
            log.info("消息已发布 msgID:{}, producer group:{}, topic:{}, tag:{}", eventMsg.getMsgId(), producerGroupName, eventMsg.getTopic(), eventMsg.getTag());
        } catch (Exception e) {
            log.error("向RocketMQ发布消息失败：" + e.getMessage(), e);
            e.printStackTrace();
        }
    }
    
    
    @PostConstruct
    public void init() {
        producer = new DefaultMQProducer(getAclRPCHook());
        producer.setProducerGroup(producerGroupName);
        producer.setNamesrvAddr(mqurl);
        try {
            producer.start();
            log.info("RocketMQ客户端producer初始化...");
        } catch (MQClientException e) {
            log.error("RocketMQ客户端producer初始化失败：" + e.getErrorMessage(), e);
        }
    }
    
    private RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
    }
}

消费者Handler接口：

/**
 *  消息消费者interface
 *  支持批量消费
 * */
public interface MQMsgHandler {
    public void handleMsg(List<EventMessage> eventMessages);
}

消费者注解：

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface MsgConsumer {
    public String consumerGroup();
    public String topic();
    public String tag();
}

最后是将消费者Handler实现注入到Spring：

/**
 * regist a consumer to MQ for all MQMsgHandler implementations
 * 
 * */
@Slf4j
@Component
public class MQConsumersRegister implements BeanPostProcessor{
    
    @Value("${rocketmq.url}")
    private String mqurl;
    
    @Value("${rocketmq.accessKey}")
    private String accessKey;
    
    @Value("${rocketmq.secretKey}")
    private String secretKey;
    
    @Value("${rocketmq.consumeThreadCorePoolSize:20}")
    private int consumeThreadCorePoolSize;

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        MsgConsumer msgConsumer = bean.getClass().getAnnotation(MsgConsumer.class);
        if(null != msgConsumer) { //如果有@MsgConsumer注解
            String consumerGroup = msgConsumer.consumerGroup();
            String topic = msgConsumer.topic();
            String tag = msgConsumer.tag();
            MQMsgHandler msgHandler = (MQMsgHandler) bean; //当前bean应是MQMsgHandler接口的实现
            
            registConsumer(msgHandler, consumerGroup, topic, tag);
            log.info("消费者组{}已订阅主题{}下{}事件" , consumerGroup, topic , tag );
        }
        return bean;
    }
    
    //注册consumer，并使其订阅相应的topic、tag
    private void registConsumer(MQMsgHandler msgHandler, String consumerGroup, String topic, String tag) {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup, getAclRPCHook(), new AllocateMessageQueueAveragely());
        try {
            consumer.setNamesrvAddr(mqurl);
            consumer.setConsumeThreadMin(consumeThreadCorePoolSize);
            consumer.setConsumeThreadMax(consumeThreadCorePoolSize);
            consumer.setPullBatchSize(32); //一次长轮询最多从mq里拿多少个消息，默认32
            consumer.subscribe(topic, tag);
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                    ConsumeConcurrentlyContext context) {
                    
                    List<EventMessage> eventMsgs = new ArrayList<>();
                    String msgContent = null;
                    
                    try {
                        for(MessageExt msg : msgs) {
                            msgContent = new String(msg.getBody(), "utf-8");
                            EventMessage eventMsg = JSON.parseObject(msgContent, EventMessage.class);
                            log.debug(JSON.toJSONString(eventMsg));
                            eventMsgs.add(eventMsg);
                        }
                        
                        msgHandler.handleMsg(eventMsgs);    //批量处理本次拉取的消息,执行业务逻辑
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        log.error("消息编码错误：" + e.getMessage(), e);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumer.start();
        }catch(Exception e) {
            log.error("注册消费者出错" + e.getMessage(), e);
            
        }
    }
    
    //Access Control List控制
    private RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
    }

}

具体使用的话，笔者在使用OpenResty+Redis+RocketMQ构建秒杀系统 - 简书 (jianshu.com)一文里实际已经用到了这个RocketMQ小工具库。用来做订单的异步入库。这里再贴一下代码：

/**
 * 异步写预约订单记录
 * */
private void sendAppointmentToMq(AppointmentDetail appointDetail) {
    String appointJson = JSON.toJSONString(appointDetail);
    EventMessage eventMsg = new EventMessage();
    eventMsg.setTopic("order");
    eventMsg.setTag("newOrder");
    eventMsg.setMsgBody(appointJson);
    
    generalMqProducer.asyncPublish(eventMsg, new SendCallback() {

        @Override
        public void onSuccess(SendResult sendResult) {
            log.info("预约订单入库消息写入rocketmq成功，消息ID:{}", sendResult.getMsgId());
            
        }

        @Override
        public void onException(Throwable e) {
            //如果与mq通信故障了，那么可以从日志文件里找到预约记录，手工执行写入mysql
            log.error("预约订单写入rocketmq失败:{}, exception detail:{}" , appointJson , e.getMessage());
        }
        
    });
}

然后消费者是可以批量的对消息进行消费的：

@Slf4j
@Component
@MsgConsumer(consumerGroup = "newOrder-consumer-group", tag = "newOrder", topic = "order")
public class NewOrderMsgHandler implements MQMsgHandler{
    
    @Autowired
    private AppointmentDetailRepository appointmentDetailRepository;

    @Override
    public void handleMsg(List<EventMessage> eventMessages) {
        
        log.debug("收到mq消息: {}", JSON.toJSONString(eventMessages));
        
        List<AppointmentDetail> appointmentDetails = new ArrayList<>();
        for(EventMessage eventMsg : eventMessages) {
            AppointmentDetail appointmentDetail = JSON.parseObject(eventMsg.getMsgBody(), AppointmentDetail.class);
            appointmentDetails.add(appointmentDetail);
        }

        appointmentDetailRepository.saveAll(appointmentDetails); //批量入库
    }

}

总结一下，这个封装库就是典型的消息发布订阅模式：发消息用生产者工具类GeneralMqProducer的相应方法直接发就行了，订阅的话需要实现MQMsgHandler接口、在接口实现类上使用@MsgConsumer注解来声明订阅的topic和tag、重写handleMsg(eventMessages)方法来编写收到消息后的业务回调逻辑即可。