SpringCloud中Rabbitmq

1.pom配置，添加以来jar包

1 <dependency>
2     <groupId>org.springframework.cloud</groupId>
3     <artifactId>spring-cloud-starter-bus-amqp</artifactId>
4 </dependency>

2.配置文件配置

1 #RabbitMq
2 spring.rabbitmq.host=${dev.spring.rabbitmq.host}
3 spring.rabbitmq.port=${dev.spring.rabbitmq.port}
4 spring.rabbitmq.username=${dev.spring.rabbitmq.username}
5 spring.rabbitmq.password=${dev.spring.rabbitmq.password}
6
7 spring.rabbitmq.exchange=${dev.spring.rabbitmq.exchange}
8 spring.rabbitmq.routingkey=${dev.spring.rabbitmq.routingkey}
9 spring.rabbitmq.queue=${dev.spring.rabbitmq.queue}

setting.properties

3.rabbitmq消息发送一

1 @Autowired 
2 private AmqpTemplate amqpTemplate;
3 //key：消息名称；content：消息内容
4 public void send(String key,String content) {
5     amqpTemplate.convertAndSend(key, content); 
6 }

4.rabbitmq消息接收一

1 @Component
2 @EnableScheduling
3 public class ClmNeedPaperMqReciever extends ClmBaseMqReciever {
4 /**
5 *RabbitListener：rabbitmq监听；queues = "queues1"：监听的队列名*称；@Payload String orderXML：接收到的消息内容
6 */
7 @RabbitHandler
8 @RabbitListener(queues = "queues1", containerFactory = "rabbitListenerContainerFactory")
9 public void process(@Payload String orderXML) {
10
11 //处理内容
12 }

接受消息

5.rabbitmq发送二

1 String MQname ="ClmClose";
2 ConnectionFactory factory = new ConnectionFactory();
3 factory.setHost("129.1.9.193");
4 factory.setUsername("admin");
5 factory.setPassword("admin");
6 factory.setPort(5672);
7 Connection connection;
8 try {
9 connection = factory.newConnection();
10 Channel channel = connection.createChannel();
11 channel.queueDeclare(MQname, true, false, false, null);
12 channel.basicPublish("", MQname, null, returnJSON.getBytes("UTF-8"));
13 channel.close();
14 connection.close();

发送mq消息

6.rabbitmq接收二

1 /* 建立连接 */
2 ConnectionFactory factory = new ConnectionFactory();
3 factory.setHost("129.1.9.193");// MQ的IP
4 factory.setPort(5672);// MQ端口
5 factory.setUsername("admin");// MQ用户名
6 factory.setPassword("admin");// MQ密码
7 Connection connection = factory.newConnection();
8 Channel channel = connection.createChannel();
9
10 /* 声明要连接的队列 */
11 channel.queueDeclare(QUEUE_NAME, true, false, false, null);
12 System.out.println("等待消息产生：");
13
14 /* 创建消费者对象，用于读取消息 */
15 QueueingConsumer consumer = new QueueingConsumer(channel);
16 channel.basicConsume(QUEUE_NAME, true, consumer);
17
18 /* 读取队列，并且阻塞，即在读到消息之前在这里阻塞，直到等到消息，完成消息的阅读后，继续阻塞循环 */
19 while (true) {
20 QueueingConsumer.Delivery delivery = consumer.nextDelivery();
21 String message = new String(delivery.getBody(),"UTF-8");
22 System.out.println("收到消息'" + message + "'");
23 }

接受mq消息

7.用接收一的方法去接收发送二的内容时需要转换，否则会出现乱码

1 @Component
2 @EnableScheduling
3 public class ClmNeedPaperMqReciever extends ClmBaseMqReciever {
4
5 /**
6 *RabbitListener：rabbitmq监听；queues = "queues1"：监听的队
7 *列名称；@Payload String orderXML：接收到的消息内容
8 */
9 @RabbitHandler
10 @RabbitListener(queues = "queues1", containerFactory = "rabbitListenerContainerFactory")
11 public void process(@Payload String orderXML) {
12
13 //处理内容
14
15 // 解析报文对象，将byte数组还原
16 String[] listStr = orderXML.split(",");
17 byte[] listByte = new byte[listStr.length];
18 for (int i = 0; i < listStr.length; i++) {
19 listByte[i] = new Byte(listStr[i]);
20 }
21 orderXML = new String(listByte, "UTF-8");
22
23 }
24
25

接收消息