RocketMq整合记录备用

引入pom

1
2
3
4
5
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.1.1</version>
</dependency>

一, 生产者

1 配置文件

1
2
3
4
5
6
7
rocketmq:
  name-server: {mq地址}:{mq端口}  # NameServer 地址，集群使用';'隔开
  producer:
    group: producer-group  # 生产者组名称
    send-message-timeout: 3000
    retry-times-when-send-failed: 2
    retry-next-server: true

2 配置类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;

@Configuration
public class RocketMqConfiguration {

    @Value("${rocketmq.name-server}")
    private String nameServer;
    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    //配置生产者
    @Bean
    public DefaultMQProducer defaultMQProducer() {
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(nameServer);
        return producer;
    }
}

3 消费者发送服务 (可以使用工厂类,此处不示例)

接口定义发送规范

1
2
3
public interface MyRocketService {
    SendResult sendMsg(String topic, String tag, Object data);
}

实现类 发送实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import com.alibaba.fastjson.JSONObject;
import com.ren.rocketmqdemo.service.MyRocketService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class MyRocketProducer implements MyRocketService {

    @Autowired
    private DefaultMQProducer ownProducer;

    @Override
    public SendResult sendMsg(String topic, String tag, Object data) {
        // 切换为本地MQ服务
        String body = JSONObject.toJSONString(data);
        log.info("Local Ons发送消息topic:{},tag:{} msg:{}", topic, tag, JSONObject.toJSONString(data));
        Message msg = new Message(topic, tag, body.getBytes());
        SendResult sendResult = null;
        try {
            sendResult = ownProducer.send(msg);
            log.info(String.format("Local MQ消息发送状态:%s, queueId:%d, body:%s", sendResult.getSendStatus(),
                    sendResult.getMessageQueue().getQueueId(), body));
        } catch (Exception e) {
            e.printStackTrace();
            log.error("MQ消息发送失败", e);
        }
        return sendResult;
    }
}

4 使用示例

1
2
3
4
5
6
7
8
9
10
11
12
@RestController
@RequestMapping("/rocketmq")
public class RocketMqController {

    @Autowired
    private MyRocketProducer myRocketProducer;

    @GetMapping("/{msg}")
    public SendResult sendMsg(@PathVariable("msg") String msg) {
        return myRocketProducer.sendMsg("test-topic", "test-tag", msg);
    }
}

二, 消费者

1 配置文件

1
2
3
4
5
rocketmq:
  name-server: {mq地址}:{mq端口}  # NameServer 地址，集群使用';'隔开
  consumer:
    group: consumer-group  # 消费者组名称
    topic: test-topic  # 订阅的主题

两种监听方式

2-1 注解形式

首先在启动类加上RocketMq配置引入注解

1
2
3
4
5
6
7
@Import(RocketMQAutoConfiguration.class)
public class Application {

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

监听业务类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Service;

@Service
@RocketMQMessageListener(topic = "test-topic", selectorExpression = "test-tag || demo", consumerGroup ="${rocketmq.consumer.group}" ,selectorType = SelectorType.TAG)  // selectorExpression指定要过滤的tag
public class MyRocketCustomer implements RocketMQListener<String> {

    @Override
    public void onMessage(String message) {
        System.out.println(message);
        // 处理接收到的消息
        // 方法体
    }
}

2-2 方法体形式 (不推荐)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@Value("${rocketmq.name-server}")
private String namesrv;

// 此处传入的mqHandler是一个工厂类方法,通过公用父类MqHandler动态传入子类,多态的利用消息做出不同响应
@Override
public void registerOrderListener(String consumerGroups, String topic, String tag, MqHandler handler) {
	try {
		String mapKey = consumerGroups + topic + tag;
		if (consumerMap.containsKey(mapKey)) {
			throw new SystemException(GlobalError.MQ_CONSUMER_REPEAT_ERROR.code,
					GlobalError.MQ_CONSUMER_REPEAT_ERROR.msg);
		}
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroups, true);
		consumer.setNamesrvAddr(namesrv);
		/**
		 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
		 * 如果非第一次启动，那么按照上次消费的位置继续消费
		 */
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		consumer.subscribe(topic, tag);
		consumer.registerMessageListener(new MessageListenerOrderly() {
			@Override
			public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
				context.setAutoCommit(true);
				for (MessageExt message : msgs) {
					long time = System.currentTimeMillis();
					String key = message.getKeys();
					String tag = message.getTags();
					String msgid = message.getMsgId();
					String content = null;
					try {
						content = new String(message.getBody(), "UTF-8");
					} catch (UnsupportedEncodingException e) {
						log.info("MQ注册消费者失败：{}", e.getMessage());
						throw new SystemException(GlobalError.MQ_CONSUMER_ERROR.code,
								GlobalError.MQ_CONSUMER_ERROR.msg);
					}
					log.info("MQ消息key:{} tag:{} msgid:{} content:{}", key, tag, msgid, content);
                       // 工厂类动态多态方法
					MqStatus status = handler.methodT(content);
					if (!MqStatus.SUCCESS.equals(status)) {
						log.info("MQ消息key:{} 消费结束:{}，耗时：{}ms", key, status, (System.currentTimeMillis() - time));
						throw new SystemException(GlobalError.MQ_CONSUMER_ERROR.code,
								GlobalError.MQ_CONSUMER_ERROR.msg);
					}
				}
				return ConsumeOrderlyStatus.SUCCESS;
			}
		});
		consumer.start();
		consumerMap.put(mapKey, consumer);
	} catch (MQClientException e) {
		log.info("MQ注册消费者失败：{}", e.getErrorMessage());
		throw new SystemException(GlobalError.MQ_CONSUMER_INIT_ERROR.code, GlobalError.MQ_CONSUMER_INIT_ERROR.msg);
	}

}