rocketMQ-封装库使用

一，引入本地依赖，注意自己的电脑是否配置有本地库，没配置问一下别人

     <!--<dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.3.0</version>
     </dependency>-->  这个就不要了，关于 rocketmq的一切依赖都可以删掉，只需要保留以下本地库

        <dependency>
            <groupId>org.dxstudio</groupId>
            <artifactId>dx-rocketmq-starter</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>

二，配置文件增加以下内容

environment: local  #这里各个环境要使用不同的环境标记，叫什么不重要，目前统一 local | dev | pron | prod
rocketmq:
  consumer:
    # 一次拉取消息最大值，注意是拉取消息的最大值而非消费最大值
    pull-batch-size: 10
  producer:
    group: dx-dump-group
    # 发送消息超时时间，默认3000
    sendMessageTimeout: 3000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2
    # 异步消息重试此处，默认2
    retryTimesWhenSendAsyncFailed: 2
    # 消息最大长度，默认1024 * 1024 * 4(默认4M)
    maxMessageSize: 4096
    # 压缩消息阈值，默认4k(1024 * 4)
    compressMessageBodyThreshold: 4096
    # 是否在内部发送失败时重试另一个broker，默认false
    retryNextServer: false
  name-server: 192.168.222.10:9876

三，main上添加注解

为了自动注入三方库中的类。因为在配置中做了注解拦截和修改，用于自动隔离环境，自动更改你的侦听器里的group和topic

@MapperScan("org.dxstudio.dump.core.mapper")
@EnableAsync
@SpringBootApplication
@ComponentScan("org.dxstudio.*")  //加上这个
public class DumpApplication {

    public static void main(String[] args) {
        SpringApplication.run(DumpApplication.class, args);
    }

}

四，发布消息

已经使用项目组的规则做了封装，为保持统一，虽然没有二次封装 rocketMQTemplate，也请使用封装好的 topic 和message 来

@Autowired
RocketMQTemplate rocketMQTemplate;

private SendResult sendMsg(int taskId ){
        //第一个参数是标签，我们约定为事件类型，
        //第二个参数是消息的key（用于在mq后台查询），这里如果是订单 直接订单号，其他使用ID即可
        //第三个参数是你要发送的消息体，范型，爱传什么传什么
        DxMQTpl<Integer> t = new DxMQTpl<>("task", String.valueOf( taskId ), taskId);

        //没有再二次封装 rocketMQTemplate， 所以可以网上搜索这个的发送方法即可，没有特别情况一般就用 syncSend
        SendResult result = rocketMQTemplate.syncSend( t.topic(), t.message() );
        log.info("发送消息 {}", result) ;
        return result;

}

五，消费消息

消费消息已经封装好了处理方法，继承DxMQMessageHandler<T>、并实现 RocketMQListener< DxMQMessage<T> > 即可。

@RocketMQMessageListener 这个是 rocketMQTemplate 的注释，可网上查资料，一般就用我下面的即可，改成自己要订阅的频道

@Slf4j
@RocketMQMessageListener(consumerGroup = "dx-dump-group",
        topic = "dx-dump",
        selectorExpression = "task",
        selectorType = SelectorType.TAG,
        consumeThreadMax = 1,
        consumeThreadNumber = 1)
public class Consumer extends DxMQMessageHandler<Long> implements RocketMQListener< DxMQMessage<Long> > {
    @Autowired
    TaskScanService taskScanService;
    
    //这里固定写 执行超类方法即可
    @Override
    public void onMessage( DxMQMessage<Long> msg) {
        super.dispatchMessage( msg );
    }
    
    //这里是自己真正要处理的逻辑，有问题毫无犹豫抛出异常即可
    @Override
    protected void handleMessage(DxMQMessage<Long> message) throws Exception {
        log.info("处理消息 {}", message);
//        throw new RuntimeException("测试一下异常");
    }
    
    //这里是异常时要做的事情，比如标记异常。 
    //这里必须是重试都失败了才会最后触发，比如关闭 isRetry 和 throwException。 throwException开启会触发rocketmq的自动重发
    @Override
    protected void handleMaxRetriesExceeded(DxMQMessage<Long> message) {
        log.error("消费失败，后续处理，比如标记异常");
    }
    

    //过滤方法，返回true不会执行handleMessage。 用于限制重复的消息消费(rocketmq不会保证不重复)，比如订单已经处理过了，在这里个方法里检查好，就不应该再继续处理
    @Override
    protected boolean filter(DxMQMessage<Long> message) {
        return super.filter(message);
    }
    
    //如果要自己修改重试规则(mq的重试是一直会有)，开启true，则每次发生异常，是由代码控制 重新发消息。这样自己可以控制重试多少次(5秒间隔)，达到最大次数后触发handleMaxRetriesExceeded
    //这个要看自己的业务，允许异常的业务 用这个，不允许异常的，使用mq的重试
    @Override
    protected boolean isRetry() {
        return true;
    }

    @Override
    protected boolean throwException() {
        return true;
    }
}

六，附上rocketmq的重发机制

Joson

rocketMQ-封装库使用

joson