[feature]添加基于 JMS 的 ActiveMQ 实例

boyunkai · boyunkai · commit ad46466504aa · 2021-02-05T15:56:28.000+08:00
diff --git a/09mq/activemq-demo/src/main/java/io/byk/activemq/JmsActiveMqApplication.java b/09mq/activemq-demo/src/main/java/io/byk/activemq/JmsActiveMqApplication.java
@@ -2,6 +2,7 @@
 
 
 import static io.byk.activemq.config.ActiveMqConfig.ACTIVE_MQ_QUEUE;
+import static io.byk.activemq.config.ActiveMqConfig.ACTIVE_MQ_TOPIC;
 
 import java.util.Queue;
 
@@ -13,6 +14,7 @@
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
 import io.byk.activemq.queue.QueueProducer;
+import io.byk.activemq.topic.TopicPublisher;
 import lombok.extern.slf4j.Slf4j;
 
 /**
@@ -24,6 +26,8 @@
 public class JmsActiveMqApplication implements ApplicationRunner {
     @Resource
     private QueueProducer queueProducer;
+    @Resource
+    private TopicPublisher topicPublisher;
 
     public static void main(String[] args) {
         SpringApplication.run(JmsActiveMqApplication.class, args);
@@ -32,9 +36,13 @@ public static void main(String[] args) {
     @Override
     public void run(ApplicationArguments args) throws Exception {
         for (int i = 0; i < 10; i++) {
-            String message;
-            log.info(message = "生产消息" + i);
+            String message = "队列消息" + i;
             queueProducer.sendMessage(ACTIVE_MQ_QUEUE, message);
         }
+
+        for (int i = 0; i < 10; i++) {
+            String message = "主题消息" + i;
+            topicPublisher.sendMessage(ACTIVE_MQ_TOPIC, message);
+        }
     }
 }
diff --git a/09mq/activemq-demo/src/main/java/io/byk/activemq/topic/JmsContainerConfig.java b/09mq/activemq-demo/src/main/java/io/byk/activemq/topic/JmsContainerConfig.java
@@ -0,0 +1,26 @@
+package io.byk.activemq.topic;
+
+import javax.jms.ConnectionFactory;
+
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jms.config.JmsListenerContainerFactory;
+import org.springframework.jms.config.SimpleJmsListenerContainerFactory;
+
+/**
+ * @author boyunkai <boyunkai@kuaishou.com>
+ * Created on 2021-02-05
+ */
+@Configuration
+public class JmsContainerConfig {
+
+    @Bean
+    public JmsListenerContainerFactory<?> myJmsContainerFactory(@Qualifier("jmsConnectionFactory")
+            ConnectionFactory connectionFactory) {
+        SimpleJmsListenerContainerFactory factory = new SimpleJmsListenerContainerFactory();
+        factory.setConnectionFactory(connectionFactory);
+        factory.setPubSubDomain(true);
+        return factory;
+    }
+}
diff --git a/09mq/activemq-demo/src/main/java/io/byk/activemq/topic/TopicPublisher.java b/09mq/activemq-demo/src/main/java/io/byk/activemq/topic/TopicPublisher.java
@@ -0,0 +1,33 @@
+package io.byk.activemq.topic;
+
+import javax.annotation.Resource;
+import javax.jms.Destination;
+
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.springframework.jms.core.JmsMessagingTemplate;
+import org.springframework.stereotype.Service;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 队列生产者
+ * @author boyunkai <boyunkai@kuaishou.com>
+ * Created on 2021-02-05
+ */
+@Service
+@Slf4j
+@Getter
+public class TopicPublisher {
+    // JMS 模板
+    @Resource
+    private JmsMessagingTemplate jmsMessagingTemplate;
+
+    public void sendMessage(String destinationName, String message) {
+        log.info("============> 发送 Topic 消息" + message);
+        Destination destination = new ActiveMQTopic(destinationName);
+        jmsMessagingTemplate.convertAndSend(destination, message);
+    }
+
+}
diff --git a/09mq/activemq-demo/src/main/java/io/byk/activemq/topic/TopicSubscriber.java b/09mq/activemq-demo/src/main/java/io/byk/activemq/topic/TopicSubscriber.java
@@ -0,0 +1,24 @@
+package io.byk.activemq.topic;
+
+import static io.byk.activemq.config.ActiveMqConfig.ACTIVE_MQ_QUEUE;
+import static io.byk.activemq.config.ActiveMqConfig.ACTIVE_MQ_TOPIC;
+
+import org.springframework.jms.annotation.JmsListener;
+import org.springframework.stereotype.Service;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 队列消费者
+ *
+ * @author boyunkai <boyunkai@kuaishou.com>
+ * Created on 2021-02-05
+ */
+@Service
+@Slf4j
+public class TopicSubscriber {
+    @JmsListener(destination = ACTIVE_MQ_TOPIC,containerFactory = "myJmsContainerFactory")
+    public void receiveMessage(String message) {
+        log.info("<=========== 收到消息" + message);
+    }
+}
