添加代码示例springboot-rabbitmq

springboot-rabbitmq/src/main/java/com/xncoding/pos/Application.java

@@ -0,0 +1,12 @@
+package com.xncoding.pos;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+
+@SpringBootApplication
+public class Application {
+    public static void main(String[] args) {
+        SpringApplication.run(Application.class, args);
+    }
+
+}

springboot-rabbitmq/src/main/java/com/xncoding/pos/config/RabbitConfig.java

@@ -0,0 +1,152 @@
+package com.xncoding.pos.config;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.*;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.annotation.Resource;
+
+/**
+ * RabbitConfig
+ *
+ * @author XiongNeng
+ * @version 1.0
+ * @since 2018/3/1
+ */
+@Configuration
+public class RabbitConfig {
+    @Resource
+    private RabbitTemplate rabbitTemplate;
+
+    /**
+     * 定制化amqp模版      可根据需要定制多个
+     * <p>
+     * <p>
+     * 此处为模版类定义 Jackson消息转换器
+     * ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调   即消息发送到exchange  ack
+     * ReturnCallback接口用于实现消息发送到RabbitMQ 交换器，但无相应队列与交换器绑定时的回调  即消息发送不到任何一个队列中  ack
+     *
+     * @return the amqp template
+     */
+    // @Primary
+    @Bean
+    public AmqpTemplate amqpTemplate() {
+        Logger log = LoggerFactory.getLogger(RabbitTemplate.class);
+        // 使用jackson 消息转换器
+        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
+        rabbitTemplate.setEncoding("UTF-8");
+        // 消息发送失败返回到队列中，yml需要配置 publisher-returns: true
+        rabbitTemplate.setMandatory(true);
+        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
+            String correlationId = message.getMessageProperties().getCorrelationIdString();
+            log.debug("消息：{} 发送失败, 应答码：{} 原因：{} 交换机: {}  路由键: {}", correlationId, replyCode, replyText, exchange, routingKey);
+        });
+        // 消息确认，yml需要配置 publisher-confirms: true
+        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
+            if (ack) {
+                log.debug("消息发送到exchange成功,id: {}", correlationData.getId());
+            } else {
+                log.debug("消息发送到exchange失败,原因: {}", cause);
+            }
+        });
+        return rabbitTemplate;
+    }
+
+    /* ----------------------------------------------------------------------------Direct exchange test--------------------------------------------------------------------------- */
+
+    /**
+     * 声明Direct交换机 支持持久化.
+     *
+     * @return the exchange
+     */
+    @Bean("directExchange")
+    public Exchange directExchange() {
+        return ExchangeBuilder.directExchange("DIRECT_EXCHANGE").durable(true).build();
+    }
+
+    /**
+     * 声明一个队列 支持持久化.
+     *
+     * @return the queue
+     */
+    @Bean("directQueue")
+    public Queue directQueue() {
+        return QueueBuilder.durable("DIRECT_QUEUE").build();
+    }
+
+    /**
+     * 通过绑定键 将指定队列绑定到一个指定的交换机 .
+     *
+     * @param queue    the queue
+     * @param exchange the exchange
+     * @return the binding
+     */
+    @Bean
+    public Binding directBinding(@Qualifier("directQueue") Queue queue,
+                                 @Qualifier("directExchange") Exchange exchange) {
+        return BindingBuilder.bind(queue).to(exchange).with("DIRECT_ROUTING_KEY").noargs();
+    }
+
+    /* ----------------------------------------------------------------------------Fanout exchange test--------------------------------------------------------------------------- */
+
+    /**
+     * 声明 fanout 交换机.
+     *
+     * @return the exchange
+     */
+    @Bean("fanoutExchange")
+    public FanoutExchange fanoutExchange() {
+        return (FanoutExchange) ExchangeBuilder.fanoutExchange("FANOUT_EXCHANGE").durable(true).build();
+    }
+
+    /**
+     * Fanout queue A.
+     *
+     * @return the queue
+     */
+    @Bean("fanoutQueueA")
+    public Queue fanoutQueueA() {
+        return QueueBuilder.durable("FANOUT_QUEUE_A").build();
+    }
+
+    /**
+     * Fanout queue B .
+     *
+     * @return the queue
+     */
+    @Bean("fanoutQueueB")
+    public Queue fanoutQueueB() {
+        return QueueBuilder.durable("FANOUT_QUEUE_B").build();
+    }
+
+    /**
+     * 绑定队列A 到Fanout 交换机.
+     *
+     * @param queue          the queue
+     * @param fanoutExchange the fanout exchange
+     * @return the binding
+     */
+    @Bean
+    public Binding bindingA(@Qualifier("fanoutQueueA") Queue queue,
+                            @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
+        return BindingBuilder.bind(queue).to(fanoutExchange);
+    }
+
+    /**
+     * 绑定队列B 到Fanout 交换机.
+     *
+     * @param queue          the queue
+     * @param fanoutExchange the fanout exchange
+     * @return the binding
+     */
+    @Bean
+    public Binding bindingB(@Qualifier("fanoutQueueB") Queue queue,
+                            @Qualifier("fanoutExchange") FanoutExchange fanoutExchange) {
+        return BindingBuilder.bind(queue).to(fanoutExchange);
+    }
+}

springboot-rabbitmq/src/main/java/com/xncoding/pos/mq/Receiver.java

@@ -0,0 +1,61 @@
+package com.xncoding.pos.mq;
+
+import com.rabbitmq.client.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/**
+ * 消息监听器
+ *
+ * @author XiongNeng
+ * @version 1.0
+ * @since 2018/3/1
+ */
+@Component
+public class Receiver {
+    private static final Logger log = LoggerFactory.getLogger(Receiver.class);
+
+    /**
+     * FANOUT广播队列监听一.
+     *
+     * @param message the message
+     * @param channel the channel
+     * @throws IOException the io exception  这里异常需要处理
+     */
+    @RabbitListener(queues = {"FANOUT_QUEUE_A"})
+    public void on(Message message, Channel channel) throws IOException {
+        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
+        log.debug("FANOUT_QUEUE_A " + new String(message.getBody()));
+    }
+
+    /**
+     * FANOUT广播队列监听二.
+     *
+     * @param message the message
+     * @param channel the channel
+     * @throws IOException the io exception   这里异常需要处理
+     */
+    @RabbitListener(queues = {"FANOUT_QUEUE_B"})
+    public void t(Message message, Channel channel) throws IOException {
+        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
+        log.debug("FANOUT_QUEUE_B " + new String(message.getBody()));
+    }
+
+    /**
+     * DIRECT模式.
+     *
+     * @param message the message
+     * @param channel the channel
+     * @throws IOException the io exception  这里异常需要处理
+     */
+    @RabbitListener(queues = {"DIRECT_QUEUE"})
+    public void message(Message message, Channel channel) throws IOException {
+        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
+        log.debug("DIRECT " + new String(message.getBody()));
+    }
+}

springboot-rabbitmq/src/main/java/com/xncoding/pos/service/SenderService.java

@@ -0,0 +1,43 @@
+package com.xncoding.pos.service;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.CorrelationData;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.UUID;
+
+/**
+ * 消息发送服务
+ */
+@Service
+public class SenderService {
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+    @Resource
+    private RabbitTemplate rabbitTemplate;
+
+    /**
+     * 测试广播模式.
+     *
+     * @param p the p
+     * @return the response entity
+     */
+    public void broadcast(String p) {
+        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
+        rabbitTemplate.convertAndSend("FANOUT_EXCHANGE", "", p, correlationData);
+    }
+
+    /**
+     * 测试Direct模式.
+     *
+     * @param p the p
+     * @return the response entity
+     */
+    public void direct(String p) {
+        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
+        rabbitTemplate.convertAndSend("DIRECT_EXCHANGE", "DIRECT_ROUTING_KEY", p, correlationData);
+    }
+
+}

springboot-rabbitmq/src/main/resources/application.yml

@@ -0,0 +1,38 @@
+##########################################################
+##################  所有profile共有的配置  #################
+##########################################################
+
+###################  spring配置  ###################
+spring:
+  profiles:
+    active: dev
+
+---
+
+#####################################################################
+########################  开发环境profile  ##########################
+#####################################################################
+spring:
+  profiles: dev
+  rabbitmq:
+    host: 123.207.66.156
+    port: 5672
+    username: spring
+    password: 123456
+    publisher-confirms: true #支持发布确认
+    publisher-returns: true  #支持发布返回
+    listener:
+      simple:
+        acknowledge-mode: manual #采用手动应答
+        concurrency: 1 #指定最小的消费者数量
+        max-concurrency: 1 #指定最大的消费者数量
+        retry:
+          enabled: true #是否支持重试
+
+logging:
+  level:
+    ROOT: INFO
+    com:
+      xncoding: DEBUG
+  file: E:/logs/app.log
+

springboot-rabbitmq/src/test/java/com/xncoding/service/SenderServiceTest.java

@@ -0,0 +1,35 @@
+package com.xncoding.service;
+
+import com.xncoding.pos.Application;
+import com.xncoding.pos.service.SenderService;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+
+/**
+ * UserServiceTest
+ *
+ * @author XiongNeng
+ * @version 1.0
+ * @since 2018/2/2
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = Application.class)
+public class SenderServiceTest {
+    @Autowired
+    private SenderService senderService;
+
+    @Test
+    public void testCache() {
+        // 测试广播模式
+        senderService.broadcast("同学们集合啦！");
+        // 测试Direct模式
+        senderService.direct("定点消息");
+    }
+}