MQ組件重磅更新，靈活切換Rocket/Redis/Kafka/Rabbit多種實現
作者：Jensen
關于上不上MQ，團隊內贊同者還是占多數的，只要能解決現有的問題，讓程序猿開發起來爽一些不好么？最后我們討論多次后采用了相對權衡的方案——用Redis的發布訂閱模型去實現MQ。
哈嘍，各位代碼戰士們，我是Jensen，一個夢想著和大家一起在代碼的海洋里遨游，順便撿起那些散落的知識點的程序員小伙伴。
筆者進公司一年多，一直想上MQ（消息隊列）很久了，但這邊的技術相對保守，不是因為MQ不好，要看團隊適不適合，團隊里總有一些聲音不想上MQ：
- MQ占用資源太大了，客戶的服務器資源不足，這么多微服務，16G的內存已經扛不住了
- 上了MQ誰來維護？運維也是有成本的，別到時開發用著很爽，運維就一堆負擔
- 不用MQ，系統調用是同步的，被調用方掛了，調用方直接能感知到，業務做重試就行了，不會有亂七八糟的問題
- 還有，怎么處理分布式事務問題？
確實是一些比較現實繞不開的話題，但上MQ也有非常多的好處，這里我說兩點：
- 系統解耦：微服務各個系統是有不同定位的，讓一個底層系統去往業務系統調就很不合理。比如我們有個租賃業務的微服務，調用鏈路就不應該由基礎平臺服務去調用租賃服務，而應該由租賃去監聽基礎平臺的事件，依賴反轉嘛，SpringIOC也是這個道理
- 職責劃分：通過MQ解耦后，往往開發人員的職責邊界也更清晰。只需要把MQ事件發布出去就行了，我管你消費者是哪個接口呢？我就非常受不了這邊的IOT服務，想把當前的IOT設備的在線狀態同步出去業務系統，既調用基礎平臺的同步設備狀態接口，又調用租賃設備的同步設備狀態接口，萬一后續又多了個業務要做設備的，是不是還得IOT再調一次？還有商城的店鋪同步功能也一樣，需要在Nacos配置哪個租戶需要調哪個系統的同步店鋪接口，多開一個租戶還得配置一下，不然都不知道為什么店鋪沒有同步出去。
關于上不上MQ，團隊內贊同者還是占多數的，只要能解決現有的問題，讓程序猿開發起來爽一些不好么？最后我們討論多次后采用了相對權衡的方案——用Redis的發布訂閱模型去實現MQ。
那怎么去實現這個MQ呢？實現Redis的發布訂閱模型本身是非常簡單的，AI分分鐘能幫你寫好一個Redis工具類，但MQ的范疇更大，需要一定的設計模式，例如SpringStream是能做到不更改代碼而換MQ實現的，我們自己去實現，也要參考這個思想。
之前在我的D3Boot框架內也寫過一個base-mq組件，當時只有kafka實現，現在看來有點雞肋了，得升級一下，所以我花了幾天功夫，把主流的幾個MQ的共通點抽了出來，再全部實現了一遍，話不多說，來看看MQ組件升級后是怎么用的。
一、新MQ組件的打開方式
我們來舉一個例子：商城服務在創建店鋪后，需要發MQ事件，由IOT服務和租賃服務處理后續的邏輯，那么MQ的生產者是商城，MQ的消費者分別是IOT、租賃。
首先定義一個店鋪同步事件ShopSyncEvent：
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class ShopSyncEvent extends MQEvent {
// 店鋪ID
private String shopId;
// 店鋪名稱
private String shopName;
// 店鋪詳情
private JSONObject shopInfo;
@Override
public String getTopic() {
return "mall";
}
@Override
public String getTag() {
return "shop_sync";
}
}MQ生產者（商城服務）發布MQ事件：
// 在創建店鋪成功后，發布店鋪同步MQ事件
ShopSyncEvent.builder().shopId(shopInfo.getId()).shopName(shopInfo.getName())
.shopInfo(new JSONObject(shopInfo)).build().tenantId(shopInfo.getTenantId())
.publish();MQ消費者（IOT服務、租賃服務）分別監聽MQ事件：
@Slf4j
@Component
public class MallMQListener {
/**
* 店鋪同步
*/
@MQEventListener(topic = "mall", tags = "shop_sync")
public void onShopSync(ShopSyncEvent event) {
log.info("店鋪同步至租賃/IOT：{}", event.getShopInfo());
// TODO 具體的消費邏輯
}
}那么問題來了，具體的MQ在框架內是如何實現的？好學的你此時選擇了繼續往下看~
二、定義核心組件
首先我們使用門面模式去設計一個自定義注解@MQEventListener，該注解與具體的MQ實現無關，是最輕量的存在：
/**
* MQ事件監聽器，配合MQEvent使用
*/
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface MQEventListener {
// 消費者組，不設置默認為當前啟動的應用名${spring.application.name}，一般多實例同一個消費者需要保持一致，避免同一業務重復消費
String group() default "";
// 主題，大分類，消費線程隔離，用于區分不同業務。也可作為小分類使用
String topic() default "DEFAULT";
// 標簽列表，小分類，同一主題下共享消費線程，支持：通配符*、或||、非-，支持復合表達式如：A || B -C
String tags() default "*";
}再設計一個基類MQEvent，發送與接收的MQ事件統一繼承該基類：
/**
* MQ事件基類
*/
@Data
public class MQEvent implements Serializable {
// 消息ID，默認當前時間戳
protected String msgId;
// 主題，配置base-mq.default-topic后無須每次指定
protected String topic;
// 標簽，只支持單個標簽，多標簽需要分開發送
protected String tag;
// 租戶ID，默認從線程上下文獲取
protected String tenantId;
// 發布MQ事件
public void publish() {
this.publish(getTopic(), getTag(), getTenantId());
}
// 發布MQ事件
public void publish(String topic) {
this.publish(topic, getTag(), getTenantId());
}
// 發布MQ事件
public void publish(String topic, String tag) {
this.publish(topic, tag, getTenantId());
}
// 發布MQ事件
public void publish(String topic, String tag, String tenantId) {
setTopic(topic);
if (getTopic() == null) {
setTopic(SpringContext.getEnv().getProperty("base-mq.default-topic", "DEFAULT"));
}
setTag(tag);
setTenantId(tenantId != null ? tenantId : ThreadContext.get(ContextConstants.TENANT_ID));
setMsgId(getMsgId() == null ? String.valueOf(System.currentTimeMillis()) : getMsgId());
// 這里的BaseContext是在基礎框架全局使用的上下文，用于解耦各個組件具體的技術代碼
if (BaseContext.contains("MQEventPublisher")) {
BaseContext.<String, Consumer<MQEvent>>get("MQEventPublisher").accept(this);
}
}
public <T extends MQEvent> T tenantId(String tenantId) {
this.tenantId = tenantId;
return (T) this;
}
}接下來設計一個用于抽象不同MQ實現的配置文件BaseMQProperties：
@Data
@ConfigurationProperties(prefix = "base-mq")
public class BaseMQProperties {
// 是否啟用
private boolean enable = false;
// MQ實現：目前支持kafka|rocket|redis|rabbit，一個應用只用一套MQ發布和訂閱
private String impl = "none";
// 服務地址
private String server = "";
// 命名空間，或作為所有主題的前綴，用于環境隔離等場景，如UAT/生產/租戶環境共用一個MQ
private String namespace = "";
// 是否持久化到本地，需要定義實現了MQEventStorer的Bean
private boolean persist = false;
// 序列化器，定義實現了MQEventSerialization的Bean
private String serialization = "JsonMQEventSerialization";
// 是否自動提交
private boolean autoAck = false;
// 發送失敗重試次數
private int retries = 0;
// 用戶
private String username = "";
// 密碼
private String password = "";
// 生產者組
private String producerGroup = "DEFAULT";
// 默認主題
private String defaultTopic = "DEFAULT";
// 交換機
private String exchange = "";
public String namespace(String concat) {
if (this.namespace != null && !this.namespace.isEmpty()) {
return namespace + concat;
}
return "";
}
}最最重要的MQ組件MQClient，設計為一個接口，并在接口內定義default方法：
/**
* MQClient接口，MQ實現類實現該接口做差異化實現
*/
public interface MQClient {
// MQ實現
String impl();
// MQ配置
default BaseMQProperties config() {
return SpringContext.getBean(BaseMQProperties.class);
}
// 整體初始化
default void init() {
if (!config().isEnable() || !Objects.equals(config().getImpl(), impl())) return;
Logger log = LoggerFactory.getLogger("### BASE-MQ : " + impl() + "Client ###");
// 初始化MQ事件發布器
if (Objects.equals(config().getImpl(), impl()) && !BaseContext.contains("MQEventPublisher")) {
log.info("Initializing MQEventPublisher");
Consumer<MQEvent> producer = initProducer();
if (producer != null) {
BaseContext.inject("MQEventPublisher", producer);
}
}
// 初始化MQ事件監聽器
log.info("Initializing MQEventListener");
String[] beanNames = SpringContext.getBeanDefinitionNames();
List<MQListener> mqListeners = new ArrayList<>();
for (String name : beanNames) {
try {
// 找出所有Bean下注解了@MQEventListener的方法
Class<?> targetClass = AopUtils.getTargetClass(SpringContext.getBean(name));
Method[] declaredMethods = targetClass.getDeclaredMethods();
for (Method listenerMethod : declaredMethods) {
MQEventListener mqEventListener = AnnotationUtils.findAnnotation(listenerMethod, MQEventListener.class);
if (mqEventListener != null) {
// 消費者組，不設置默認按Spring工程的應用名隔離，當然也可以自定義
String group = mqEventListener.group();
if (group.isEmpty()) {
group = SpringContext.getEnv().getProperty("spring.application.name");
}
// 取出首個參數，用于接收MQ事件后自動解析到該類型
Class<MQEvent> eventClass = (Class<MQEvent>) listenerMethod.getParameterTypes()[0];
mqListeners.add(MQListener.builder()
.listenerMethod(listenerMethod)
.eventClass(eventClass)
.consumer(mqEvent -> {
try {
// 接收并解析為MQEvent后，具體調用該方法的位置
listenerMethod.invoke(SpringContext.getBean(listenerMethod.getDeclaringClass()), mqEvent);
} catch (Exception e) {
throw new RuntimeException(e);
}
})
.group(group)
.topic(mqEventListener.topic())
.tags(mqEventListener.tags())
.build());
}
}
} catch (Exception ignore) {
}
}
for (MQListener listener : mqListeners) {
try {
// 具體創建消費者的邏輯，不同MQ實現不一樣
initConsumer(listener);
listener.setSuccess(true);
} catch (Exception e) {
log.error("Listen MQ [{}] failed!", listener.topicAndTags(), e);
}
}
log.info("Listening MQ: {}", mqListeners.stream().filter(MQListener::isSuccess).map(MQListener::topicAndTags).collect(Collectors.joining(",")));
}
// 初始化生產者
Consumer<MQEvent> initProducer();
// 初始化消費者
void initConsumer(MQListener mqListener) throws Exception;
// 啟動
void start();
// 序列化器，默認取配置了的實現了MQEventSerialization的Bean
default MQEventSerialization serialization() {
return SpringContext.getBean(config().getSerialization());
}
// 消費MQ事件
default void consume(MQListener listener, MQEvent mqEvent) {
Logger log = LoggerFactory.getLogger("### BASE-MQ : " + impl() + "Client ###");
log.info("Consume MQ: {}", serialization().<String>serialize(mqEvent));
// MQ事件持久化
if (config().isPersist()) {
try {
MQEventStorer mqEventStorer = SpringContext.getBean(MQEventStorer.class);
mqEventStorer.store(mqEvent.getTopic(), mqEvent);
} catch (Exception e) {
log.error("Persist MQ failed: {}", serialization().serialize(mqEvent), e);
}
}
// 再調用方法執行業務
listener.getConsumer().accept(mqEvent);
}
}最后咱們定義好配置類，統一加載那些實現了MQClient的Bean：
/**
* 基礎MQ配置
*/
@Order(PriorityOrdered.HIGHEST_PRECEDENCE + 10)
@Configuration
@ConditionalOnProperty(prefix = "base-mq", name = "enable", havingValue = "true")
public class BaseMQConfig {
@Autowired
private List<MQClient> mqClients;
@PostConstruct
public void init() {
if (mqClients != null && !mqClients.isEmpty()) {
// 獲取所有實現了MQClient的MQ客戶端實現類
for (MQClient mqClient : mqClients) {
// 先初始化
mqClient.init();
// 再啟動
mqClient.start();
}
}
}
}核心組件中的其它類比較簡單，這里先略過。
三、定義不同的MQClient實現
各個MQ之間的特性在這里就不一一介紹了，其中70%左右的代碼由阿里的通義千問AI實現（推薦大家平時多用用），直接看實現代碼吧，相關作用在注釋里有介紹：
1.RocketMQ實現
/**
* Rocket客戶端實現
*/
@Slf4j(topic = "### BASE-MQ : rocketClient ###")
public final class RocketClient implements MQClient {
private static Map<String, Supplier> LISTENERS = new ConcurrentHashMap<>();
@Override
public String impl() {
return "rocket";
}
@Override
public Consumer<MQEvent> initProducer() {
try {
// 創建 DefaultMQProducer 實例
DefaultMQProducer rocketProducer;
if (!config().getUsername().isEmpty() && !config().getPassword().isEmpty()) {
// 需要鑒權
rocketProducer = new DefaultMQProducer(config().getNamespace(), config().getProducerGroup(),
new AclClientRPCHook(new SessionCredentials(config().getUsername(),
config().getPassword())));
} else {
rocketProducer = new DefaultMQProducer(config().getNamespace(), config().getProducerGroup());
}
rocketProducer.setNamesrvAddr(config().getServer());
// 啟動rocketProducer實例
rocketProducer.start();
DefaultMQProducer finalRocketProducer = rocketProducer;
return mqEvent -> {
// 定義具體的MQ事件發布邏輯
String message = serialization().serialize(mqEvent);
log.info("Publish MQ: {}", message);
String tags = mqEvent.getTag() == null ? "" : mqEvent.getTag();
try {
finalRocketProducer.send(new Message(mqEvent.getTopic(), tags, message.getBytes(RemotingHelper.DEFAULT_CHARSET)));
} catch (Exception e) {
log.error("Publish MQ: {} failed!", message, e);
}
};
} catch (MQClientException e) {
log.error("創建Rocket生產者失敗", e);
}
return null;
}
@Override
public void initConsumer(MQListener mqListener) throws Exception {
DefaultMQPushConsumer defaultMQPushConsumer;
if (!config().getUsername().isEmpty() && !config().getPassword().isEmpty()) {
// 需要鑒權的情況
defaultMQPushConsumer = new DefaultMQPushConsumer(config().getNamespace(), mqListener.getGroup(),
new AclClientRPCHook(new SessionCredentials(config().getUsername(), config().getPassword())));
} else {
defaultMQPushConsumer = new DefaultMQPushConsumer(config().getNamespace(), mqListener.getGroup());
}
defaultMQPushConsumer.setNamesrvAddr(config().getServer());
defaultMQPushConsumer.subscribe(mqListener.getTopic(), mqListener.getTags());
// 創建MessageListenerConcurrently實例
defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (messageExts, context) -> {
// 獲取方法參數
for (MessageExt messageExt : messageExts) {
MQEvent mqEvent = null;
try {
// 反序列化為MQEvent對象
mqEvent = serialization().deserialize(new String(messageExt.getBody(), StandardCharsets.UTF_8), mqListener.getEventClass());
consume(mqListener, mqEvent);
} catch (Exception e) {
log.error("Consume MQ failed: {}", mqEvent, e);
if (!(e instanceof ServiceException)) {
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
LISTENERS.put(mqListener.getTopic(), () -> defaultMQPushConsumer);
}
@Override
@SneakyThrows
public void start() {
if (!config().isEnable() || !Objects.equals(config().getImpl(), impl())) return;
// 在此統一啟動所有消費者
for (Supplier<DefaultMQPushConsumer> listener : LISTENERS.values()) {
listener.get().start();
}
}
}2.Kafka實現
之前底層依賴使用spring-data-kafka，用了兩年，回過頭來發現Spring封裝得太死了，后來換成了kafka-clients：
/**
* Kafka客戶端實現
*/
@Slf4j(topic = "### BASE-MQ : kafkaClient ###")
public final class KafkaClient implements MQClient, DisposableBean {
private static List<Supplier> LISTENERS = new ArrayList<>();
@Override
public String impl() {
return "kafka";
}
@Override
public Consumer<MQEvent> initProducer() {
// 創建 Producer 配置
Properties props = new Properties();
props.put("bootstrap.servers", config().getServer()); // Kafka broker 地址
props.put("acks", "all");
props.put("retries", config().getRetries());// 失敗重試次數
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// 創建KafkaProducer實例
Producer<String, String> producer = new KafkaProducer<>(props);
return mqEvent -> {
// 定義具體的MQ事件發布邏輯
String message = serialization().serialize(mqEvent);
log.info("Publish MQ: {}", message);
try {
// topic使用下劃線的方式拼接
producer.send(new ProducerRecord<>(config().namespace("_") + mqEvent.getTopic(), message));
} catch (Exception e) {
log.error("Publish MQ: {} failed!", message, e);
}
};
}
@Override
public void initConsumer(MQListener mqListener) throws Exception {
Properties props = new Properties();
props.put("bootstrap.servers", config().getServer());
props.put("group.id", mqListener.getGroup()); // 消費者組
props.put("enable.auto.commit", config().isAutoAck());
props.put("auto.offset.reset", "earliest"); // 從最早的偏移量開始消費
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList(config().namespace("_") + mqListener.getTopic()));
LISTENERS.add(() -> {
GlobalThreadPool.submit(() -> {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, String> record : records) {
// 處理消息
// Kafka沒有標簽過濾機制，所以自己實現一遍
MQEvent mqEvent = serialization().deserialize(record.value(), mqListener.getEventClass());
if (!MQFilter.matchExps(mqEvent.getTag(), mqListener.getTags())) {
continue;
}
try {
consume(mqListener, mqEvent);
if (!config().isAutoAck()) {
consumer.commitSync(); // 手動提交偏移量
}
} catch (Exception e) {
log.error("Consume MQ failed: {}", mqEvent, e);
if (e instanceof ServiceException) {
if (!config().isAutoAck()) {
consumer.commitSync(); // 手動提交偏移量
}
}
}
}
}
});
return consumer;
});
}
@Override
public void start() {
if (!config().isEnable() || !Objects.equals(config().getImpl(), impl())) return;
LISTENERS.forEach(Supplier::get);
}
@Override
public void destroy() throws Exception {
for (Supplier<KafkaConsumer> listener : LISTENERS) {
listener.get().close();
}
}
}3.Redis實現
Redis嚴格來說不算MQ，但因為它是一個獨立于應用之外的帶發布/訂閱功能的中間件，且算它半個MQ吧，一般緩存服務用得多，如果要考慮減少服務器資源開銷，可以考慮資源重復利用：
/**
* Redis客戶端實現
*/
@Slf4j(topic = "### BASE-MQ : redisClient ###")
@Component
public final class RedisClient implements MQClient, DisposableBean {
private static JedisPool JEDIS_POOL;
private static List<Supplier> LISTENERS = new ArrayList<>();
@Override
public String impl() {
return "redis";
}
@Override
public Consumer<MQEvent> initProducer() {
return mqEvent -> {
String message = serialization().serialize(mqEvent);
log.info("Publish MQ: {}", message);
// channel=namespace:topic:tag或namespace:topic或topic
String channel = String.format("%s%s%s", (config().getNamespace().isEmpty() ? "" : config().getNamespace() + ":"),
mqEvent.getTopic(),
(mqEvent.getTag() == null || mqEvent.getTag().isEmpty()) ? "" : (":" + mqEvent.getTag()));
GlobalThreadPool.submit(() -> {
try {
jedis().publish(channel, message);
} catch (Throwable e) {
log.error("Publish MQ to [{}] failed: {}", channel, message, e);
}
});
};
}
@Override
public void initConsumer(MQListener mqListener) throws Exception {
// channel=namespace:topic:tag或namespace:topic或topic，TODO 暫不支持通配符*和非-
List<String> channels = new ArrayList<>();
if (mqListener.getTags() != null && !mqListener.getTags().isEmpty()) {
Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
if (!tags.isEmpty()) {
for (String tag : tags) {
channels.add((config().getNamespace().isEmpty() ? "" : config().getNamespace() + ":")
+ mqListener.getTopic() + (":" + tag));
}
}
} else {
channels.add((config().getNamespace().isEmpty() ? "" : config().getNamespace() + ":") + mqListener.getTopic());
}
// Redis原子加鎖腳本
String lockScript = "if redis.call('get', KEYS[1]) == false then redis.call('setex', KEYS[1], tonumber(ARGV[1]), KEYS[1]) return 1 else return 0 end";
LISTENERS.add(() -> {
GlobalThreadPool.submit(() -> {
try (Jedis jedis = jedis()) {
jedis.subscribe(new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
MQEvent event = null;
try {
event = serialization().deserialize(message, mqListener.getEventClass());
//按組消費：Redis本身沒有消費組的概念，這里我們鎖住某條消息的消費資格10秒鐘，以實現同消費組（如多實例）只能由一個消費者消費
String lockKey = String.format("ChannelGroupLock:%s:%s:%s", channel, mqListener.getGroup(), event.getMsgId());
if ((long) jedis().eval(lockScript, 1, lockKey, "10000") == 1) {
//本條消息在該組未消費
consume(mqListener, event);
}
} catch (Exception e) {
log.error("Consume MQ failed: {}", event, e);
}
}
@Override
public void onSubscribe(String channel, int subscribedChannels) {
log.info("Subscribed channel: {}", channel);
}
@Override
public void onUnsubscribe(String channel, int subscribedChannels) {
log.info("Unsubscribed channel: {}", channel);
}
}, channels.toArray(new String[0]));
}
});
return null;
});
}
@Override
public void start() {
if (!config().isEnable() || !Objects.equals(config().getImpl(), impl())) return;
LISTENERS.forEach(Supplier::get);
}
@Override
public void destroy() throws Exception {
if (JEDIS_POOL != null) {
JEDIS_POOL.close();
}
}
private Jedis jedis() {
if (JEDIS_POOL == null) {
String[] hostAndPort = config().getServer().split(":");
JedisPoolConfig poolConfig = new JedisPoolConfig();
poolConfig.setMaxTotal(100);
poolConfig.setMaxIdle(50);
poolConfig.setMinIdle(10);
poolConfig.setTestOnBorrow(true);
if (!config().getPassword().isEmpty()) {
JEDIS_POOL = new JedisPool(poolConfig, hostAndPort[0], Integer.parseInt(hostAndPort[1]), 5000, config().getPassword());
} else {
JEDIS_POOL = new JedisPool(poolConfig, hostAndPort[0], Integer.parseInt(hostAndPort[1]));
}
}
return JEDIS_POOL.getResource();
}
}之前的底層依賴使用了spring-data-redis，但我發現很多工程中RedisTemplate版本沖突太多了，一氣之下我換成更輕量的jedis（spring-data-redis底層也是用jedis）。
4.RabbitMQ實現
RabbitMQ也是常用的MQ之一，我們需要考慮兼容它的AMQP協議，以及標簽Tags的實現，這里用了它默認的Direct交換機：
@Component
@Slf4j(topic = "### BASE-MQ : rabbitClient ###")
public final class RabbitClient implements MQClient, DisposableBean {
private static Connection CONNECTION;
private static List<Supplier> LISTENERS = new ArrayList<>();
@Override
public String impl() {
return "rabbit";
}
@Override
public Consumer<MQEvent> initProducer() {
try {
com.rabbitmq.client.Channel channel = connection().createChannel();
return mqEvent -> {
// 定義具體的MQ事件發布邏輯
String message = serialization().serialize(mqEvent);
log.info("Publish MQ: {}", message);
try {
// 路由鍵=namespace.topic或namespace.topic.tag
String routingKey = config().namespace(".") + mqEvent.getTopic() + (mqEvent.getTag() == null || mqEvent.getTag().isEmpty() ? "" : ("." + mqEvent.getTag()));
channel.basicPublish(config().getExchange(), routingKey, null, message.getBytes());
} catch (Exception e) {
log.error("Publish MQ: {} failed!", message, e);
}
};
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public void initConsumer(MQListener mqListener) throws Exception {
// 隊列名=group.namespace.topic.className.methodName
String queue = mqListener.group(".") + config().namespace(".") + mqListener.getListenerMethod().getDeclaringClass().getSimpleName() + "." + mqListener.getListenerMethod().getName();
List<String> routingKeys = new ArrayList<>();
if (mqListener.getTags() != null && !mqListener.getTags().isEmpty()) {
// 通過多個tags綁定到對應的路由鍵
Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
if (!tags.isEmpty()) {
// 路由鍵=namespace.topic.tag1、namespace.topic.tag2，TODO 這里只能用或||的關系，通配符*和非-是不能生效的
for (String tag : tags) {
String routingKey = config().namespace(".") + mqListener.getTopic() + "." + tag;
routingKeys.add(routingKey);
}
}
} else {
// 路由鍵=namespace.topic
String routingKey = config().namespace(".") + mqListener.getTopic();
routingKeys.add(routingKey);
}
try (com.rabbitmq.client.Channel channel = connection().createChannel()) {
channel.queueDeclare(queue, true, false, false, null);
// 隊列綁定到多個路由器
for (String routingKey : routingKeys) {
channel.queueBind(queue, config().getExchange(), routingKey);
}
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
MQEvent event = serialization().deserialize(message, mqListener.getEventClass());
if (!MQFilter.matchExps(event.getTag(), mqListener.getTags())) {
return;
}
try {
consume(mqListener, event);
if (!config().isAutoAck()) {
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
} catch (Exception e) {
log.error("Consume MQ failed: {}", event, e);
if (e instanceof ServiceException) {
if (!config().isAutoAck()) {
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}
};
LISTENERS.add(() -> GlobalThreadPool.submit(() -> {
try {
channel.basicConsume(queue, config().isAutoAck(), deliverCallback, consumerTag -> {
});
} catch (Exception e) {
}
return channel;
})
);
}
}
@Override
public void start() {
if (!config().isEnable() || !Objects.equals(config().getImpl(), impl())) return;
LISTENERS.forEach(Supplier::get);
}
@Override
public void destroy() throws Exception {
try {
for (Supplier<com.rabbitmq.client.Channel> listener : LISTENERS) {
listener.get().close();
}
} catch (Exception e) {
log.error("Error closing RabbitMQ channel", e);
}
try {
connection().close();
} catch (Exception e) {
log.error("Error closing RabbitMQ connection", e);
}
}
private Connection connection() {
if (CONNECTION == null) {
ConnectionFactory factory = new ConnectionFactory();
String[] hostAndPort = config().getServer().split(":");
factory.setHost(hostAndPort[0]);
factory.setPort(Integer.parseInt(hostAndPort[1]));
factory.setUsername(config().getUsername());
factory.setPassword(config().getPassword());
try {
CONNECTION = factory.newConnection();
} catch (IOException | TimeoutException e) {
throw new RuntimeException(e);
}
}
return CONNECTION;
}
}責任編輯：姜華
來源：
架構師修行錄
