RabbitMQ 学习

yu.jl · yu.jl · commit d8a61d10aa82 · 2019-12-12T13:20:40.000+08:00
diff --git a/dependencies.gradle b/dependencies.gradle
@@ -16,6 +16,7 @@ ext.libraries = [
         "spring-boot-starter-aop"            : "org.springframework.boot:spring-boot-starter-aop:${versions.springBoot}",
         "spring-boot-starter-log4j2"         : "org.springframework.boot:spring-boot-starter-log4j2:${versions.springBoot}",
         "spring-boot-starter-test"           : "org.springframework.boot:spring-boot-starter-test:${versions.springBoot}",
+        "spring-boot-starter-amqp"           : "org.springframework.boot:spring-boot-starter-amqp:${versions.springBoot}",
         "spring-boot-starter-tomcat"         : "org.springframework.boot:spring-boot-starter-tomcat:${versions.springBoot}",
         "spring-boot-starter-jdbc"           : "org.springframework.boot:spring-boot-starter-jdbc:${versions.springBoot}",
         "spring-boot-starter-data-rest"      : "org.springframework.boot:spring-boot-starter-data-rest:${versions.springBoot}",
diff --git a/settings.gradle b/settings.gradle
@@ -14,5 +14,6 @@ include 'StudyAdminZuul'
 include 'studyadmin'
 include 'studyapi'
 include 'common'
-include 'studyconfig2'
+include 'studystreamrmq'
+include 'studystreamrmqcustomer'
 
diff --git a/studystreamrmq/build.gradle b/studystreamrmq/build.gradle
@@ -0,0 +1,9 @@
+
+version '1.0-SNAPSHOT'
+
+
+dependencies {
+    implementation libraries['spring-boot-starter-amqp']
+    implementation libraries['fastjson']
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+}
diff --git a/studystreamrmq/src/main/java/com/ityu/studystreamrmq/rabbitmq/RabbitMqConfig.java b/studystreamrmq/src/main/java/com/ityu/studystreamrmq/rabbitmq/RabbitMqConfig.java
@@ -1,10 +1,12 @@
 package com.ityu.studystreamrmq.rabbitmq;
 
 
+import org.springframework.amqp.ImmediateAcknowledgeAmqpException;
 import org.springframework.amqp.core.*;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -89,6 +91,8 @@ Binding orderTtlBinding(DirectExchange orderTtlDirect, Queue orderTtlQueue) {
                 .with(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey());
     }
 
+
+    //https://www.jianshu.com/p/2c5eebfd0e95
 //    @Bean
 //    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
 //        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
@@ -98,5 +102,23 @@ Binding orderTtlBinding(DirectExchange orderTtlDirect, Queue orderTtlQueue) {
 //        return factory;
 //    }
 
+//    @Bean
+//    public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory) {
+//        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+//        container.setConnectionFactory(connectionFactory);
+//        container.setQueueNames("consumer_queue");              // 监听的队列
+//        container.setAcknowledgeMode(AcknowledgeMode.AUTO);     // 根据情况确认消息
+//        container.setMessageListener((MessageListener) (message) -> {
+//            System.out.println("====接收到消息=====");
+//            System.out.println(new String(message.getBody()));
+//            //抛出NullPointerException异常则重新入队列
+//            //throw new NullPointerException("消息消费失败");
+//            //当抛出的异常是AmqpRejectAndDontRequeueException异常的时候，则消息会被拒绝，且requeue=false
+//            //throw new AmqpRejectAndDontRequeueException("消息消费失败");
+//            //当抛出ImmediateAcknowledgeAmqpException异常，则消费者会被确认
+//            throw new ImmediateAcknowledgeAmqpException("消息消费失败");
+//        });
+//        return container;
+//    }
 
 }
diff --git a/studystreamrmq/src/test/java/com/ityu/studystreamrmq/MyTest.java b/studystreamrmq/src/test/java/com/ityu/studystreamrmq/MyTest.java
@@ -0,0 +1,41 @@
+package com.ityu.studystreamrmq;
+
+
+import com.ityu.studystreamrmq.rabbitmq.QueueEnum;
+import com.ityu.studystreamrmq.rabbitmq.TestSendMessage;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class MyTest {
+
+    @Autowired
+    private TestSendMessage amqpTemplate;
+    @Autowired
+    private RabbitTemplate amqpTemplate2;
+
+    @Test
+    public void sendMsg() {
+        amqpTemplate.sendMsg(QueueEnum.QUEUE_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_ORDER_CANCEL.getName(), "你好啊不错哦");
+    }
+
+    @Test
+    public void sendMsg2() {
+        System.out.println(System.currentTimeMillis());
+        amqpTemplate2.convertAndSend(QueueEnum.QUEUE_TTL_ORDER_CANCEL.getExchange(), QueueEnum.QUEUE_TTL_ORDER_CANCEL.getRouteKey(), "我是延迟消息你能发现不50000", message -> {
+            message.getMessageProperties().setExpiration("500000");
+            return message;
+        });
+    }
+
+
+}
