業(yè)務(wù)背景

目前支付中心的支付回調(diào)消息是由支付中心發(fā)送消息給各自的業(yè)務(wù)方，而且是只發(fā)送給業(yè)務(wù)方。

但是有些其他的業(yè)務(wù)也需要收到支付回調(diào)的話，目前是需要設(shè)置不同的消費(fèi)者名稱去接收支付回調(diào)。但是這樣會(huì)有點(diǎn)問題，因?yàn)槠渌麡I(yè)務(wù)可能需要同時(shí)接收多種業(yè)務(wù)的支付回調(diào)，這樣的話就需要注冊很多的消費(fèi)者，或者說多種業(yè)務(wù)想要聚合成一種處理方式，去接收多個(gè)topic消息就不太好。

所以采用支付回調(diào)時(shí)轉(zhuǎn)發(fā)的方案，發(fā)送支付回調(diào)時(shí)，如果需要轉(zhuǎn)發(fā)，那么發(fā)送支付回調(diào)的同時(shí)，發(fā)送消息給轉(zhuǎn)發(fā)服務(wù)，由轉(zhuǎn)發(fā)服務(wù)去決定發(fā)送給誰，其他業(yè)務(wù)方也只需要配置轉(zhuǎn)發(fā)規(guī)則就行。

轉(zhuǎn)發(fā)設(shè)計(jì)

橋接消息

我們可以往大了想一個(gè)概念一一橋接。轉(zhuǎn)發(fā)消息，其實(shí)也是橋接消息的一個(gè)子類。

配置

我們首先需要有一個(gè)配置表，配置這個(gè)橋接是屬于什么類型的。

CREATE TABLE `bus_msg_bridge_config` (
  `id` varchar(50) NOT NULL,
  `type` varchar(20) DEFAULT NULL,
  `create_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

這張表就是橋接配置表，type字段我們約定好一個(gè)枚舉

public enum BridgeMsgType implements MsgType {
   /**
    * 轉(zhuǎn)發(fā)
    */
   FORWARD("01");
}

轉(zhuǎn)發(fā)topic

有了配置表之后，我們就需要針對一個(gè)配置需要轉(zhuǎn)發(fā)出去的消息topic，配置和轉(zhuǎn)發(fā)topic是一對多的關(guān)系。

CREATE TABLE `bus_msg_bridge_forward` (
  `id` varchar(50) NOT NULL,
  `config_id` varchar(50) DEFAULT NULL COMMENT '配置id',
  `topic` varchar(255) DEFAULT NULL COMMENT '轉(zhuǎn)發(fā)topic',
  `msg_type` varchar(255) DEFAULT NULL COMMENT '消息類型',
  `create_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

這里的msg_type就是消息轉(zhuǎn)發(fā)類型，目前支持三種全部/主/備

轉(zhuǎn)發(fā)規(guī)則

CREATE TABLE `bus_msg_bridge_rule` (
  `id` varchar(50) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `rule_type` varchar(255) DEFAULT NULL COMMENT '規(guī)則類型',
  `fields` varchar(255) DEFAULT NULL COMMENT '字段',
  `val` varchar(255) DEFAULT NULL COMMENT '值',
  `display_num` int(11) DEFAULT NULL,
  `create_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

匹配規(guī)則

我們執(zhí)行轉(zhuǎn)發(fā)topic的時(shí)候，需要指定一些規(guī)則，比如說如果收到的消息內(nèi)容里面的字段name=測試2，如果規(guī)則設(shè)置的name=測試才進(jìn)行轉(zhuǎn)發(fā)，那么這條消息就會(huì)被過濾掉，不會(huì)進(jìn)行轉(zhuǎn)發(fā)的操作，這些就是屬于匹配的概念。多個(gè)字段過濾的話就需要設(shè)置多條規(guī)則，只要有一條匹配不過就不進(jìn)行轉(zhuǎn)發(fā)

這里會(huì)有個(gè)規(guī)則類型的字段，它是一個(gè)枚舉。

public enum RuleType {
   /**
    * 匹配
    */
   MATCH("01");
}

因?yàn)橐?guī)則可以配置多條，所以我們有一個(gè)中間表來存儲(chǔ)轉(zhuǎn)發(fā)topic和規(guī)則的映射表

CREATE TABLE `bus_msg_bridge_forward_rule` (
  `id` varchar(50) NOT NULL,
  `forward_id` varchar(50) DEFAULT NULL COMMENT '轉(zhuǎn)發(fā)id',
  `rule_id` varchar(50) DEFAULT NULL COMMENT '規(guī)則id',
  `create_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

橋接配置管理

配置管理后臺(tái)

新增配置

點(diǎn)擊上面的新增按鈕，新增配置，這里的id要和外層傳進(jìn)來的id一致

新增配置

即約定好外層傳進(jìn)來的id，和這里的配置id一樣即可。

新增topic轉(zhuǎn)發(fā)

點(diǎn)擊列表的右側(cè)操作按鈕添加topic

添加topic

注意事項(xiàng)：

• topic需要寫全稱
• 延遲只支持秒
• 校驗(yàn)規(guī)則多選，則需要全部通過才進(jìn)行轉(zhuǎn)發(fā)

查看topic

一個(gè)配置支持轉(zhuǎn)發(fā)到多個(gè)topic

topic查看

這里可以看到所配置的topic信息并進(jìn)行編輯

轉(zhuǎn)發(fā)實(shí)戰(zhàn)

配置轉(zhuǎn)發(fā)規(guī)則

上面的表新增數(shù)據(jù)

發(fā)送轉(zhuǎn)發(fā)消息

BridgeMsgDTO bridgeMsgDTO = new BridgeMsgDTO();
// 這個(gè)id就是上面轉(zhuǎn)發(fā)的配置id
bridgeMsgDTO.setMsgId(notifyMessageDTO.getPcNotifyConfig().getId());
bridgeMsgDTO.setContent(message);
ProducerSendMessageFacade.sendPulsarFullMsg("cp-comm/msg-daemon/bridge-msg",JSON.toJSONString(bridgeMsgDTO));

往topic(轉(zhuǎn)發(fā)的消息消費(fèi)者)發(fā)消息即可。