目录
一、添加依赖
二、在application.yml 或者 application.properties文件中进行RabbitMQ的相关配置
三、确认Rabbitmq是启动的并且是能够连接上的
四、简单测试(代码)
五、进阶测试(代码)
闲言碎语不要讲,安排
一、添加依赖
二、在配置文件对Rabbitmq进行配置
yaml格式
properti格式:
三、确认Rabbitmq是启动的并且是能够连接上的
首先,确保你的RabbitMQ已经安装好,并且已经启动,在你的本地确实可以连接的上(关于RabbitMQ的安装和相关配置在《rabbitmq集群搭建》进行设置,具体请自行查阅,谢谢。
四、简单测试(这是一对一的演示 另外一对多 多对多就不在这演示了)
这里说的一对多是指一个生产者对应多个消费者 多对多是多个生产者对应多个消费者
(1)配置队列:
package com.hh.demo;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @Description:队列配置,队列的名称,发送者和接受者的名称必须一致,否则接收不到消息
*/
@Configuration
public class RabbitMqConfig {
@Bean
public Queue Queue1() {
return new Queue("hhTest1");
}
}
(2)发送者通过Controller类发送消息:
package com.hh.demo.controller;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Date;
@RestController
public class SendController {
@Autowired
private AmqpTemplate amqpTemplate;
@RequestMapping("/send")
public String send(){
String content="Date:"+new Date();
amqpTemplate.convertAndSend("hhTest1",content);
return content;
}
}
(3)创建接受者Receiver1,新建类:
package com.hh.demo.Receiver;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "hhTest1")
public class Receiver1 {
@RabbitHandler
public void receiver(String msg){
System.out.println("Test1 receiver1:"+msg);
}
}
(4)测试
浏览器访问地址:http://localhost:8080/send,如下图:
终端输出接受的内容:
接下来演示下:
topic ExChange示例
topic 是RabbitMQ中最灵活的一种方式,可以根据binding_key自由的绑定不同的队列
首先对topic规则配置,这里使用两个队列来测试(也就是在Application类中创建和绑定的topic.message和topic.messages两个队列),其中topic.message的bindting_key为
“topic.message”,topic.messages的binding_key为“topic.#”;
生产者:
package com.rabbit.topic;
import java.util.Date;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class TopicSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String msg1 = "I am topic.mesaage msg======";
System.out.println("sender1 : " + msg1);
this.rabbitTemplate.convertAndSend("exchange", "topic.message", msg1);
String msg2 = "I am topic.mesaages msg########";
System.out.println("sender2 : " + msg2);
this.rabbitTemplate.convertAndSend("exchange", "topic.messages", msg2);
}
}
消费者1(topic.message)
package com.rabbit.topic;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "topic.message")
public class topicMessageReceiver {
@RabbitHandler
public void process(String msg) {
System.out.println("topicMessageReceiver : " +msg);
}
}
消费者2(topic.messages)
package com.rabbit.topic;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "topic.messages")
public class topicMessagesReceiver {
@RabbitHandler
public void process(String msg) {
System.out.println("topicMessagesReceiver : " +msg);
}
}
controller:
/**
* topic exchange类型rabbitmq测试
*/
@PostMapping("/topicTest")
public void topicTest() {
topicSender.send();
}
用post方式执行:
http://127.0.0.1:8080/rabbit/topicTest
结果如下:
sender1 : I am topic.mesaage msg======
sender2 : I am topic.mesaages msg########
topicMessageReceiver : I am topic.mesaage msg======
topicMessagesReceiver : I am topic.mesaage msg======
topicMessagesReceiver : I am topic.mesaages msg########
由以上结果可知:sender1发送的消息,routing_key是“topic.message”,所以exchange里面的绑定的binding_key是“topic.message”,topic.#都符合路由规则;所以sender1
发送的消息,两个队列都能接收到;
sender2发送的消息,routing_key是“topic.messages”,所以exchange里面的绑定的binding_key只有topic.#都符合路由规则;所以sender2发送的消息只有队列
topic.messages能收到。
fanout ExChange示例
Fanout 就是我们熟悉的广播模式或者订阅模式,给Fanout转发器发送消息,绑定了这个转发器的所有队列都收到这个消息。
这里使用三个队列来测试(也就是在Application类中创建和绑定的fanout.A、fanout.B、fanout.C)这三个队列都和Application中创建的fanoutExchange转发器绑定。
生产者:
package com.rabbit.fanout;
import java.util.Date;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class FanoutSender {
@Autowired
private AmqpTemplate rabbitTemplate;
public void send() {
String msgString="fanoutSender :hello i am hh";
System.out.println(msgString);
this.rabbitTemplate.convertAndSend("fanoutExchange","abcd.ee", msgString);
}
}
消费者A:
package com.rabbit.fanout;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
@RabbitHandler
public void process(String msg) {
System.out.println("FanoutReceiverA : " + msg);
}
}
消费者b:
package com.rabbit.fanout;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB {
@RabbitHandler
public void process(String msg) {
System.out.println("FanoutReceiverB : " + msg);
}
}
消费者c:
package com.rabbit.fanout;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
@RabbitHandler
public void process(String msg) {
System.out.println("FanoutReceiverC : " + msg);
}
}
controller:
/**
* fanout exchange类型rabbitmq测试
*/
@PostMapping("/fanoutTest")
public void fanoutTest() {
fanoutSender.send();
}
用post方式执行:
http://127.0.0.1:8080/rabbit/fanoutTest
结果如下:
fanoutSender :hello i am hzb
FanoutReceiverC : fanoutSender :hello i am hh
FanoutReceiverB : fanoutSender :hello i am hh
FanoutReceiverA : fanoutSender :hello i am hh
由以上结果可知:就算fanoutSender发送消息的时候,指定了routing_key为"abcd.ee",但是所有接收者都接受到了消息
带callback的消息发送
增加回调处理,这里不再使用application.properties默认配置的方式,会在程序中显示的使用文件中的配置信息。该示例中没有新建队列和exchange,用的是第5节中的topic.messages队列和exchange转发器。消费者也是第5节中的topicMessagesReceiver
rabbitmq配置类:
package com.rabbit.callback;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Scope;
public class RabbitConfig {
@Value("${spring.rabbitmq.host}")
private String addresses;
@Value("${spring.rabbitmq.port}")
private String port;
@Value("${spring.rabbitmq.username}")
private String username;
@Value("${spring.rabbitmq.password}")
private String password;
@Value("${spring.rabbitmq.virtual-host}")
private String virtualHost;
@Value("${spring.rabbitmq.publisher-confirms}")
private boolean publisherConfirms;
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
connectionFactory.setAddresses(addresses+":"+port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost(virtualHost);
/** 如果要进行消息回调,则这里必须要设置为true */
connectionFactory.setPublisherConfirms(publisherConfirms);
return connectionFactory;
}
@Bean
/** 因为要设置回调类,所以应是prototype类型,如果是singleton类型,则回调类为最后一次设置 */
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplatenew() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
return template;
}
}
生产者:
package com.rabbit.callback;
import java.util.Date;
import java.util.UUID;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class CallBackSender implements RabbitTemplate.ConfirmCallback{
@Autowired
private RabbitTemplate rabbitTemplatenew;
public void send() {
rabbitTemplatenew.setConfirmCallback(this);
String msg="callbackSender : i am callback sender";
System.out.println(msg );
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
System.out.println("callbackSender UUID: " + correlationData.getId());
this.rabbitTemplatenew.convertAndSend("exchange", "topic.messages", msg, correlationData);
}
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
// TODO Auto-generated method stub
System.out.println("callbakck confirm: " + correlationData.getId());
}
}
消费者:第5节中的topicMessagesReceiver
controller:
@PostMapping("/callback")
public void callbak() {
callBackSender.send();
}
用post方式执行:
http://127.0.0.1:8080/rabbit/callback
结果如下:
callbackSender : i am callback sender
callbackSender UUID: cd0c80a6-4c65-4bf9-b4f8-f3b1180755d6
callbakck confirm: cd0c80a6-4c65-4bf9-b4f8-f3b1180755d6
topicMessagesReceiver : callbackSender : i am callback sender
从上面可以看出callbackSender发出的UUID,收到了回应,又传回来了。