Sender的實現：

@ComponentpublicclassSenderTopic{@AutowiredprivateRabbitTemplate rabbitTemplate;/*queue的key，用于和routing key 根據binding模式匹配*/@Bean(name="message")publicQueuequeueMessage() {returnnewQueue("topic.message"); }@Bean(name="messages")publicQueuequeueMessages() {returnnewQueue("topic.messages"); }@BeanpublicTopicExchangeexchange() {returnnewTopicExchange("exchange"); }/*設置binding key,此時所有發送到這個exchange的消息， exchange都會根據routing key將消息與@Qualifier定義的queue進行匹配*/@BeanBinding bindingExchangeMessage(@Qualifier("message") Queue queueMessage, TopicExchange exchange) {returnBindingBuilder.bind(queueMessage).to(exchange).with("topic.message"); }@BeanBinding bindingExchangeMessages(@Qualifier("messages") Queue queueMessages, TopicExchange exchange) {returnBindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一個詞,#表示零個或多個詞}publicvoidsend(String routingKey, String msg) {this.rabbitTemplate.convertAndSend("exchange",routingKey, msg); } }

Listener的實現如下：

@Configuration@RabbitListener(queues ="topic.message")//監聽器監聽指定的QueuepublicclassListenerTopic{privatestaticfinalLogger LOGGER = LoggerFactory.getLogger(ListenerTopic.class);@RabbitHandlerpublicvoidprocess(@Payload String foo) { LOGGER.info("Listener: "+ foo); } }

listener2的實現如下

@Configuration@RabbitListener(queues ="topic.messages")publicclassListenerTopic2{privatestaticfinalLogger LOGGER = LoggerFactory.getLogger(ListenerTopic2.class);@RabbitHandlerpublicvoidprocess(@Payload String foo) { LOGGER.info("Listener2: "+ foo); } }

發送topic.message會匹配到topic.#和topic.message 兩個Receiver都可以收到消息，發送topic.messages（或者top、topic等）只有topic.#可以匹配所有只有Receiver2監聽到消息。

Fanout模型@ConfigurationpublicclassSenderFanout{@AutowiredprivateRabbitTemplate rabbitTemplate;/*queue的key，用于和routing key 根據binding模式匹配*/@Bean(name="Amessage")publicQueueAMessage() {returnnewQueue("fanout.A"); }@Bean(name="Bmessage")publicQueueBMessage() {returnnewQueue("fanout.B"); }@BeanFanoutExchange fanoutExchange() {returnnewFanoutExchange("fanoutExchange");//配置廣播路由器}@BeanBinding bindingExchangeA(@Qualifier("Amessage") Queue AMessage,FanoutExchange fanoutExchange) {returnBindingBuilder.bind(AMessage).to(fanoutExchange); }@BeanBinding bindingExchangeB(@Qualifier("Bmessage") Queue BMessage, FanoutExchange fanoutExchange) {returnBindingBuilder.bind(BMessage).to(fanoutExchange); }publicvoidsend(String msg) {this.rabbitTemplate.convertAndSend("fanoutExchange","", msg); } }@Configuration@RabbitListener(queues ="fanout.A")//監聽器監聽指定的QueuepublicclassListenerFanout{privatestaticfinalLogger LOGGER = LoggerFactory.getLogger(ListenerFanout.class);@RabbitHandlerpublicvoidprocess(@Payload String foo) { LOGGER.info("Listener: "+ foo); } }

Fanout模式下，所有綁定fanout交換機的隊列，都能收到消息。

Kafka Kafka簡介

Kafka是一種分布式的，基于發布/訂閱的消息系統。主要設計目標如下：

以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數據也能保證常數時間復雜度的訪問性能。

高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條以上消息的傳輸。

支持Kafka Server間的消息分區，及分布式消費，同時保證每個Partition內的消息順序傳輸。

同時支持離線數據處理和實時數據處理。

Scale out：支持在線水平擴展。