SpringCloud中Rabbitmq的使用

Posted 2020-09-26 染颜

1.pom配置

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>

2.配置文件配置

#RabbitMq
spring.rabbitmq.host=${dev.spring.rabbitmq.host}
spring.rabbitmq.port=${dev.spring.rabbitmq.port}
spring.rabbitmq.username=${dev.spring.rabbitmq.username}
spring.rabbitmq.password=${dev.spring.rabbitmq.password}

spring.rabbitmq.exchange=${dev.spring.rabbitmq.exchange}
spring.rabbitmq.routingkey=${dev.spring.rabbitmq.routingkey}
spring.rabbitmq.queue=${dev.spring.rabbitmq.queue}

3.rabbitmq消息发送一

@Autowired
private AmqpTemplate amqpTemplate;

public void send(String key,String content) {
amqpTemplate.convertAndSend(key, content);
}

key：消息名称；content：消息内容

4.rabbitmq消息接收一

@Component
@EnableScheduling
public class ClmNeedPaperMqReciever extends ClmBaseMqReciever {

@RabbitHandler
@RabbitListener(queues = "queues1", containerFactory = "rabbitListenerContainerFactory")
public void process(@Payload String orderXML) {

//处理内容
}

RabbitListener：rabbitmq监听；queues = "queues1"：监听的队列名称；@Payload String orderXML：接收到的消息内容

5.rabbitmq发送二

String MQname ="ClmClose";
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("129.1.9.193");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(5672);
Connection connection;
try {
connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(MQname, true, false, false, null);
channel.basicPublish("", MQname, null, returnJSON.getBytes("UTF-8"));
channel.close();
connection.close();

6.rabbitmq接收二

/* 建立连接 */
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("129.1.9.193");// MQ的IP
factory.setPort(5672);// MQ端口
factory.setUsername("admin");// MQ用户名
factory.setPassword("admin");// MQ密码
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

/* 声明要连接的队列 */
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
System.out.println("等待消息产生：");

/* 创建消费者对象，用于读取消息 */
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(QUEUE_NAME, true, consumer);

/* 读取队列，并且阻塞，即在读到消息之前在这里阻塞，直到等到消息，完成消息的阅读后，继续阻塞循环 */
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody(),"UTF-8");
System.out.println("收到消息‘" + message + "‘");
}

 

7.用接收一的方法去接收发送二的内容

@Component
@EnableScheduling
public class ClmNeedPaperMqReciever extends ClmBaseMqReciever {

@RabbitHandler
@RabbitListener(queues = "queues1", containerFactory = "rabbitListenerContainerFactory")
public void process(@Payload String orderXML) {

//处理内容

// 解析报文对象，将byte数组还原
String[] listStr = orderXML.split(",");
byte[] listByte = new byte[listStr.length];
for (int i = 0; i < listStr.length; i++) {
listByte[i] = new Byte(listStr[i]);
}
orderXML = new String(listByte, "UTF-8");

}

RabbitListener：rabbitmq监听；queues = "queues1"：监听的队列名称；@Payload String orderXML：接收到的消息内容