MQ組件重磅更新，靈活切換Rocket/Redis/Kafka/Rabbit多種實現(xiàn)
作者：Jensen
關(guān)于上不上MQ，團隊內(nèi)贊同者還是占多數(shù)的，只要能解決現(xiàn)有的問題，讓程序猿開發(fā)起來爽一些不好么？最后我們討論多次后采用了相對權(quán)衡的方案——用Redis的發(fā)布訂閱模型去實現(xiàn)MQ。
哈嘍，各位代碼戰(zhàn)士們，我是Jensen，一個夢想著和大家一起在代碼的海洋里遨游，順便撿起那些散落的知識點的程序員小伙伴。
筆者進公司一年多，一直想上MQ（消息隊列）很久了，但這邊的技術(shù)相對保守，不是因為MQ不好，要看團隊適不適合，團隊里總有一些聲音不想上MQ：
- MQ占用資源太大了，客戶的服務(wù)器資源不足，這么多微服務(wù)，16G的內(nèi)存已經(jīng)扛不住了
- 上了MQ誰來維護？運維也是有成本的，別到時開發(fā)用著很爽，運維就一堆負擔(dān)
- 不用MQ，系統(tǒng)調(diào)用是同步的，被調(diào)用方掛了，調(diào)用方直接能感知到，業(yè)務(wù)做重試就行了，不會有亂七八糟的問題
- 還有，怎么處理分布式事務(wù)問題？
確實是一些比較現(xiàn)實繞不開的話題，但上MQ也有非常多的好處，這里我說兩點：
- 系統(tǒng)解耦：微服務(wù)各個系統(tǒng)是有不同定位的，讓一個底層系統(tǒng)去往業(yè)務(wù)系統(tǒng)調(diào)就很不合理。比如我們有個租賃業(yè)務(wù)的微服務(wù)，調(diào)用鏈路就不應(yīng)該由基礎(chǔ)平臺服務(wù)去調(diào)用租賃服務(wù)，而應(yīng)該由租賃去監(jiān)聽基礎(chǔ)平臺的事件，依賴反轉(zhuǎn)嘛，SpringIOC也是這個道理
- 職責(zé)劃分：通過MQ解耦后，往往開發(fā)人員的職責(zé)邊界也更清晰。只需要把MQ事件發(fā)布出去就行了，我管你消費者是哪個接口呢？我就非常受不了這邊的IOT服務(wù)，想把當(dāng)前的IOT設(shè)備的在線狀態(tài)同步出去業(yè)務(wù)系統(tǒng)，既調(diào)用基礎(chǔ)平臺的同步設(shè)備狀態(tài)接口，又調(diào)用租賃設(shè)備的同步設(shè)備狀態(tài)接口，萬一后續(xù)又多了個業(yè)務(wù)要做設(shè)備的，是不是還得IOT再調(diào)一次？還有商城的店鋪同步功能也一樣，需要在Nacos配置哪個租戶需要調(diào)哪個系統(tǒng)的同步店鋪接口，多開一個租戶還得配置一下，不然都不知道為什么店鋪沒有同步出去。
關(guān)于上不上MQ，團隊內(nèi)贊同者還是占多數(shù)的，只要能解決現(xiàn)有的問題，讓程序猿開發(fā)起來爽一些不好么？最后我們討論多次后采用了相對權(quán)衡的方案——用Redis的發(fā)布訂閱模型去實現(xiàn)MQ。
那怎么去實現(xiàn)這個MQ呢？實現(xiàn)Redis的發(fā)布訂閱模型本身是非常簡單的，AI分分鐘能幫你寫好一個Redis工具類，但MQ的范疇更大，需要一定的設(shè)計模式，例如SpringStream是能做到不更改代碼而換MQ實現(xiàn)的，我們自己去實現(xiàn)，也要參考這個思想。
之前在我的D3Boot框架內(nèi)也寫過一個base-mq組件，當(dāng)時只有kafka實現(xiàn)，現(xiàn)在看來有點雞肋了，得升級一下，所以我花了幾天功夫，把主流的幾個MQ的共通點抽了出來，再全部實現(xiàn)了一遍，話不多說，來看看MQ組件升級后是怎么用的。
一、新MQ組件的打開方式
我們來舉一個例子：商城服務(wù)在創(chuàng)建店鋪后，需要發(fā)MQ事件，由IOT服務(wù)和租賃服務(wù)處理后續(xù)的邏輯，那么MQ的生產(chǎn)者是商城，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生產(chǎn)者（商城服務(wù)）發(fā)布MQ事件：
// 在創(chuàng)建店鋪成功后，發(fā)布店鋪同步MQ事件
ShopSyncEvent.builder().shopId(shopInfo.getId()).shopName(shopInfo.getName())
.shopInfo(new JSONObject(shopInfo)).build().tenantId(shopInfo.getTenantId())
.publish();MQ消費者（IOT服務(wù)、租賃服務(wù)）分別監(jiān)聽MQ事件：
@Slf4j
@Component
public class MallMQListener {
/**
* 店鋪同步
*/
@MQEventListener(topic = "mall", tags = "shop_sync")
public void onShopSync(ShopSyncEvent event) {
log.info("店鋪同步至租賃/IOT：{}", event.getShopInfo());
// TODO 具體的消費邏輯
}
}那么問題來了，具體的MQ在框架內(nèi)是如何實現(xiàn)的？好學(xué)的你此時選擇了繼續(xù)往下看~
二、定義核心組件
首先我們使用門面模式去設(shè)計一個自定義注解@MQEventListener，該注解與具體的MQ實現(xiàn)無關(guān)，是最輕量的存在：
/**
* MQ事件監(jiān)聽器，配合MQEvent使用
*/
@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface MQEventListener {
// 消費者組，不設(shè)置默認為當(dāng)前啟動的應(yīng)用名${spring.application.name}，一般多實例同一個消費者需要保持一致，避免同一業(yè)務(wù)重復(fù)消費
String group() default "";
// 主題，大分類，消費線程隔離，用于區(qū)分不同業(yè)務(wù)。也可作為小分類使用
String topic() default "DEFAULT";
// 標(biāo)簽列表，小分類，同一主題下共享消費線程，支持：通配符*、或||、非-，支持復(fù)合表達式如：A || B -C
String tags() default "*";
}再設(shè)計一個基類MQEvent，發(fā)送與接收的MQ事件統(tǒng)一繼承該基類：
/**
* MQ事件基類
*/
@Data
public class MQEvent implements Serializable {
// 消息ID，默認當(dāng)前時間戳
protected String msgId;
// 主題，配置base-mq.default-topic后無須每次指定
protected String topic;
// 標(biāo)簽，只支持單個標(biāo)簽，多標(biāo)簽需要分開發(fā)送
protected String tag;
// 租戶ID，默認從線程上下文獲取
protected String tenantId;
// 發(fā)布MQ事件
public void publish() {
this.publish(getTopic(), getTag(), getTenantId());
}
// 發(fā)布MQ事件
public void publish(String topic) {
this.publish(topic, getTag(), getTenantId());
}
// 發(fā)布MQ事件
public void publish(String topic, String tag) {
this.publish(topic, tag, getTenantId());
}
// 發(fā)布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是在基礎(chǔ)框架全局使用的上下文，用于解耦各個組件具體的技術(shù)代碼
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;
}
}接下來設(shè)計一個用于抽象不同MQ實現(xiàn)的配置文件BaseMQProperties：
@Data
@ConfigurationProperties(prefix = "base-mq")
public class BaseMQProperties {
// 是否啟用
private boolean enable = false;
// MQ實現(xiàn)：目前支持kafka|rocket|redis|rabbit，一個應(yīng)用只用一套MQ發(fā)布和訂閱
private String impl = "none";
// 服務(wù)地址
private String server = "";
// 命名空間，或作為所有主題的前綴，用于環(huán)境隔離等場景，如UAT/生產(chǎn)/租戶環(huán)境共用一個MQ
private String namespace = "";
// 是否持久化到本地，需要定義實現(xiàn)了MQEventStorer的Bean
private boolean persist = false;
// 序列化器，定義實現(xiàn)了MQEventSerialization的Bean
private String serialization = "JsonMQEventSerialization";
// 是否自動提交
private boolean autoAck = false;
// 發(fā)送失敗重試次數(shù)
private int retries = 0;
// 用戶
private String username = "";
// 密碼
private String password = "";
// 生產(chǎn)者組
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，設(shè)計為一個接口，并在接口內(nèi)定義default方法：
/**
* MQClient接口，MQ實現(xiàn)類實現(xiàn)該接口做差異化實現(xiàn)
*/
public interface MQClient {
// MQ實現(xiàn)
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事件發(fā)布器
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事件監(jiān)聽器
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) {
// 消費者組，不設(shè)置默認按Spring工程的應(yīng)用名隔離，當(dāng)然也可以自定義
String group = mqEventListener.group();
if (group.isEmpty()) {
group = SpringContext.getEnv().getProperty("spring.application.name");
}
// 取出首個參數(shù)，用于接收MQ事件后自動解析到該類型
Class<MQEvent> eventClass = (Class<MQEvent>) listenerMethod.getParameterTypes()[0];
mqListeners.add(MQListener.builder()
.listenerMethod(listenerMethod)
.eventClass(eventClass)
.consumer(mqEvent -> {
try {
// 接收并解析為MQEvent后，具體調(diào)用該方法的位置
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 {
// 具體創(chuàng)建消費者的邏輯，不同MQ實現(xiàn)不一樣
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(",")));
}
// 初始化生產(chǎn)者
Consumer<MQEvent> initProducer();
// 初始化消費者
void initConsumer(MQListener mqListener) throws Exception;
// 啟動
void start();
// 序列化器，默認取配置了的實現(xiàn)了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);
}
}
// 再調(diào)用方法執(zhí)行業(yè)務(wù)
listener.getConsumer().accept(mqEvent);
}
}最后咱們定義好配置類，統(tǒng)一加載那些實現(xiàn)了MQClient的Bean：
/**
* 基礎(chǔ)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()) {
// 獲取所有實現(xiàn)了MQClient的MQ客戶端實現(xiàn)類
for (MQClient mqClient : mqClients) {
// 先初始化
mqClient.init();
// 再啟動
mqClient.start();
}
}
}
}核心組件中的其它類比較簡單，這里先略過。
三、定義不同的MQClient實現(xiàn)
各個MQ之間的特性在這里就不一一介紹了，其中70%左右的代碼由阿里的通義千問AI實現(xiàn)（推薦大家平時多用用），直接看實現(xiàn)代碼吧，相關(guān)作用在注釋里有介紹：
1.RocketMQ實現(xiàn)
/**
* Rocket客戶端實現(xiàn)
*/
@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 {
// 創(chuàng)建 DefaultMQProducer 實例
DefaultMQProducer rocketProducer;
if (!config().getUsername().isEmpty() && !config().getPassword().isEmpty()) {
// 需要鑒權(quán)
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事件發(fā)布邏輯
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("創(chuàng)建Rocket生產(chǎn)者失敗", e);
}
return null;
}
@Override
public void initConsumer(MQListener mqListener) throws Exception {
DefaultMQPushConsumer defaultMQPushConsumer;
if (!config().getUsername().isEmpty() && !config().getPassword().isEmpty()) {
// 需要鑒權(quán)的情況
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());
// 創(chuàng)建MessageListenerConcurrently實例
defaultMQPushConsumer.registerMessageListener((MessageListenerConcurrently) (messageExts, context) -> {
// 獲取方法參數(shù)
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;
// 在此統(tǒng)一啟動所有消費者
for (Supplier<DefaultMQPushConsumer> listener : LISTENERS.values()) {
listener.get().start();
}
}
}2.Kafka實現(xiàn)
之前底層依賴使用spring-data-kafka，用了兩年，回過頭來發(fā)現(xiàn)Spring封裝得太死了，后來換成了kafka-clients：
/**
* Kafka客戶端實現(xiàn)
*/
@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() {
// 創(chuàng)建 Producer 配置
Properties props = new Properties();
props.put("bootstrap.servers", config().getServer()); // Kafka broker 地址
props.put("acks", "all");
props.put("retries", config().getRetries());// 失敗重試次數(shù)
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");
// 創(chuàng)建KafkaProducer實例
Producer<String, String> producer = new KafkaProducer<>(props);
return mqEvent -> {
// 定義具體的MQ事件發(fā)布邏輯
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沒有標(biāo)簽過濾機制，所以自己實現(xiàn)一遍
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實現(xiàn)
Redis嚴格來說不算MQ，但因為它是一個獨立于應(yīng)用之外的帶發(fā)布/訂閱功能的中間件，且算它半個MQ吧，一般緩存服務(wù)用得多，如果要考慮減少服務(wù)器資源開銷，可以考慮資源重復(fù)利用：
/**
* Redis客戶端實現(xiàn)
*/
@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秒鐘，以實現(xiàn)同消費組（如多實例）只能由一個消費者消費
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，但我發(fā)現(xiàn)很多工程中RedisTemplate版本沖突太多了，一氣之下我換成更輕量的jedis（spring-data-redis底層也是用jedis）。
4.RabbitMQ實現(xiàn)
RabbitMQ也是常用的MQ之一，我們需要考慮兼容它的AMQP協(xié)議，以及標(biāo)簽Tags的實現(xiàn)，這里用了它默認的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事件發(fā)布邏輯
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綁定到對應(yīng)的路由鍵
Set<String> tags = MQFilter.findIncludes(mqListener.getTags());
if (!tags.isEmpty()) {
// 路由鍵=namespace.topic.tag1、namespace.topic.tag2，TODO 這里只能用或||的關(guān)系，通配符*和非-是不能生效的
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;
}
}責(zé)任編輯：姜華
來源：
架構(gòu)師修行錄
