增加 activemq 示例

lab-32/lab-32-activemq-demo/src/main/java/cn/iocoder/springboot/lab32/activemqdemo/consumer/Demo01Consumer.java

@@ -3,6 +3,7 @@ package cn.iocoder.springboot.lab32.activemqdemo.consumer;
 import cn.iocoder.springboot.lab32.activemqdemo.message.Demo01Message;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.jms.annotation.JmsListener;
 import org.springframework.stereotype.Component;
 
 @Component
@@ -10,9 +11,14 @@ public class Demo01Consumer {
 
     private Logger logger = LoggerFactory.getLogger(getClass());
 
-//    @JmsListener(destination = Demo01Message.QUEUE)
+    @JmsListener(destination = Demo01Message.QUEUE)
     public void onMessage(Demo01Message message) {
         logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
     }
 
+//    @JmsListener(destination = Demo01Message.QUEUE)
+//    public void onMessage(javax.jms.Message message) {
+//        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
+//    }
+
 }

lab-32/lab-32-activemq-demo/src/main/java/cn/iocoder/springboot/lab32/activemqdemo/message/Demo01Message.java

@@ -22,7 +22,7 @@ public class Demo01Message implements Serializable {
 
     @Override
     public String toString() {
-        return "ClusteringtMessage{" +
+        return "Demo01Message{" +
                 "id=" + id +
                 '}';
     }

lab-32/lab-32-activemq-demo/src/main/java/cn/iocoder/springboot/lab32/activemqdemo/producer/Demo01Producer.java

@@ -1,23 +1,38 @@
 package cn.iocoder.springboot.lab32.activemqdemo.producer;
 
 import cn.iocoder.springboot.lab32.activemqdemo.message.Demo01Message;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.jms.core.JmsMessagingTemplate;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.scheduling.annotation.AsyncResult;
 import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
+import org.springframework.util.concurrent.ListenableFuture;
 
 @Component
 public class Demo01Producer {
 
-    @Resource
-    private JmsMessagingTemplate rabbitTemplate;
+    @Autowired
+    private JmsMessagingTemplate jmsTemplate;
 
     public void syncSend(Integer id) {
         // 创建 ClusteringMessage 消息
         Demo01Message message = new Demo01Message();
         message.setId(id);
         // 同步发送消息
-        rabbitTemplate.convertAndSend(Demo01Message.QUEUE, message);
+        jmsTemplate.convertAndSend(Demo01Message.QUEUE, message);
+    }
+
+    @Async
+    public ListenableFuture<Void> asyncSend(Integer id) {
+        try {
+            // 发送消息
+            this.syncSend(id);
+            // 返回成功的 Future
+            return AsyncResult.forValue(null);
+        } catch (Throwable ex) {
+            // 返回异常的 Future
+            return AsyncResult.forExecutionException(ex);
+        }
     }
 
 }

lab-32/lab-32-activemq-demo/src/main/resources/application.yaml

@@ -5,4 +5,4 @@ spring:
     user: admin # 账号
     password: admin # 密码
     packages:
-      trust-all: true
+      trust-all: true # 可信任的反序列化包

lab-32/lab-32-activemq-demo/src/test/java/cn/iocoder/springboot/lab32/activemqdemo/producer/Demo01ProducerTest.java

@@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.concurrent.ListenableFutureCallback;
 
 import java.util.concurrent.CountDownLatch;
 
@@ -31,4 +32,26 @@ public class Demo01ProducerTest {
         new CountDownLatch(1).await();
     }
 
+    @Test
+    public void testAsyncSend() throws InterruptedException {
+        int id = (int) (System.currentTimeMillis() / 1000);
+        producer.asyncSend(id).addCallback(new ListenableFutureCallback<Void>() {
+
+            @Override
+            public void onFailure(Throwable e) {
+                logger.info("[testASyncSend][发送编号：[{}] 发送异常]]", id, e);
+            }
+
+            @Override
+            public void onSuccess(Void aVoid) {
+                logger.info("[testASyncSend][发送编号：[{}] 发送成功，发送成功]", id);
+            }
+
+        });
+        logger.info("[testASyncSend][发送编号：[{}] 调用完成]", id);
+
+        // 阻塞等待，保证消费
+        new CountDownLatch(1).await();
+    }
+
 }