rabbitmq代码
spring:
rabbitmq:
host: # 主机名
port: 5672 # 端口
virtual-host: / # 虚拟主机
username: zhangsan # 用户名
password: 1234 # 密码
=============================================
public class User implements Serializable {
private static final long serliaVersionUid = 1L;
private String name;
@Override
public String toString() {
return "User{" +
"name='" + name + '\'' +
", age=" + age +
'}';
}
private int age;
public static long getSerliaVersionUid() {
return serliaVersionUid;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
public User(String name, int age) {
this.name = name;
this.age = age;
}
}
===========================================================
生产者
@SpringBootApplication
public class PublisherApplication {
public static void main(String[] args) {
SpringApplication.run(PublisherApplication.class);
}
@Bean
public MessageConverter jsonMessageConverter(){
return new Jackson2JsonMessageConverter();
}
}
=========================================================
@SpringBootTest(classes = PublisherApplication.class)
@RunWith(SpringRunner.class)
public class PublisherTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test/*简单消息模型*/
public void simple() {
String queueName = "simple.queue";
// 消息
String message = "hello, spring amqp!";
// 发送消息
//需要先申明一个queue
rabbitTemplate.convertAndSend(queueName, message);
}
@Test/*工作消息模型*/
public void worker() {
String queueName = "simple.queue";
// 消息
String message = "hello, spring amqp!";
// 发送消息
//需要先申明一个queue
for (int i = 0; i < 50; i++) {
rabbitTemplate.convertAndSend(queueName, "发送的第" + i + "条消息:" + message);
}
}
@Test/*简单消息模型*/
public void testSendMessage() throws IOException, TimeoutException {
// 1.建立连接
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.238.128");
factory.setPort(5672);
factory.setVirtualHost("/itcast");
factory.setUsername("zhangsan");
factory.setPassword("1234");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道Channel
Channel channel = connection.createChannel();
// 3.创建队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.发送消息
String message = "hello, rabbitmq!";
channel.basicPublish("", queueName, null, message.getBytes());
System.out.println("发送消息成功：【" + message + "】");
// 5.关闭通道和连接
channel.close();
connection.close();
}
@Test/*工作消息模型*/
public void testSendWorkerMessage() throws IOException, TimeoutException {
// 1.建立连接
ConnectionFactory factory = new ConnectionFactory();
// 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
factory.setHost("192.168.238.128");
factory.setPort(5672);
factory.setVirtualHost("/itcast");
factory.setUsername("zhangsan");
factory.setPassword("1234");
// 1.2.建立连接
Connection connection = factory.newConnection();
// 2.创建通道Channel
Channel channel = connection.createChannel();
// 3.创建队列
String queueName = "simple.queue";
channel.queueDeclare(queueName, false, false, false, null);
// 4.发送消息
String message = "hello, rabbitmq!";
for (int i = 0; i < 50; i++) {
channel.basicPublish("", queueName, null, message.getBytes());
}
System.out.println("发送消息成功：【" + message + "】");
// 5.关闭通道和连接
channel.close();
connection.close();
}
/*发布/订阅*/
@Test/*Fanout广播*/
public void fanout() {
// 消息
String message = "hello,Fanout广播!";
//交换机
String fanoutName = "fanoutExchange";
// 发送消息
//需要先申明一个queue
for (int i = 0; i < 10; i++) {
rabbitTemplate.convertAndSend(fanoutName, "", message);
}
}
@Test/*Direct定向路由*/
public void direct() {
// 消息
String message = "hello,Direct定向路由!";
//交换机
String directName = "directExchange";
// 发送消息
//需要先申明一个queue
User user = new User("张三", 22);
rabbitTemplate.convertAndSend(directName, "red", user);
rabbitTemplate.convertAndSend(directName, "blue", user);
rabbitTemplate.convertAndSend(directName, "yellow", user);
}
@Test/*Topic话题*/
public void topic() {
// 消息
String message = "hello,Topic话题!";
//交换机
String topicName = "topicExchange";
// 发送消息
//需要先申明一个queue
rabbitTemplate.convertAndSend(topicName, "china.watch", message);
rabbitTemplate.convertAndSend(topicName, "japan.watch", message);
rabbitTemplate.convertAndSend(topicName, "china.niubi", message);
}
@Test/*Topic话题*/
public void topic2() {
// 消息
String message = "hello,Topic话题!";
//交换机
String topicName = "topicExchange2";
// 发送消息
//需要先申明一个queue
rabbitTemplate.setExchange(topicName);
rabbitTemplate.convertAndSend(topicName, "china.watch", message);
rabbitTemplate.convertAndSend(topicName, "japan.watch", message);
rabbitTemplate.convertAndSend(topicName, "china.niubi", message);
}
============================================================
消费者
@SpringBootApplication
public class ConsumerApplication {
public static void main(String[] args) {
SpringApplication.run(ConsumerApplication.class, args);
}
@Bean
public MessageConverter jsonMessageConverter(){
return new Jackson2JsonMessageConverter();
}
}
==========================================================
@Component
public class FanoutConfig {
/*定义声明交换机对垒,交换机绑定*/
//1定义一个交换机
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
//2.定义声明一个队列
@Bean
public Queue queue1() {
return new Queue("fanoutQueue1");
}
@Bean
public Queue queue2() {
return new Queue("fanoutQueue2");
}
//3.绑定
@Bean
public Binding builder1(FanoutExchange fanoutExchange, Queue queue1) {
return BindingBuilder.bind(queue1).to(fanoutExchange);
}
@Bean
public Binding builder2(FanoutExchange fanoutExchange, Queue queue2) {
return BindingBuilder.bind(queue2).to(fanoutExchange);
}
}
============================================================
@Component
public class ConsumerListener {
int count1 = 1;
int count2 = 1;
@RabbitListener(queues = "simple.queue")/*简单队列监听;工作队列1*/
public void simplequeue(Object msg) {
System.err.println("Q1-接收到消息：【" + msg + "】" + count1++);
}
@RabbitListener(queues = "simple.queue") /*简单队列监听;工作队列2*/
public void simplequeue2(Object msg) throws InterruptedException {
System.out.println("Q2-接收到消息：【" + msg + "】" + count2++);
Thread.sleep(20);
}
//===========================================================================
@RabbitListener(queues = "fanoutQueue1") /* fanout广播1*/
public void fanoutQueue1(Object msg) throws InterruptedException {
System.out.println(" fanout1-接收到消息：【" + msg + "】");
Thread.sleep(20);
}
@RabbitListener(queues = "fanoutQueue2") /* fanout广播2*/
public void fanoutQueue2(Object msg) throws InterruptedException {
System.out.println(" fanout2-接收到消息：【" + msg + "】");
Thread.sleep(20);
}
//===========================================================================
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue1"),
exchange = @Exchange(name = "directExchange", type = ExchangeTypes.DIRECT),
key = {"red", "blue"}
))/*Direct定向路由1*/
public void listenDirectQueue1(Object msg) {
System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "direct.queue2"),
exchange = @Exchange(name = "directExchange", type = ExchangeTypes.DIRECT),
key = {"red", "yellow"}
))/*Direct定向路由2*/
public void listenDirectQueue2(Object msg) {
System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
}
//===========================================================================
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue2"),
exchange = @Exchange(name = "topicExchange", type = ExchangeTypes.TOPIC),
key = {"china.#"}
))/*Topic话题*/
public void listenTopicQueue2(Object msg) {
System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue1"),
exchange = @Exchange(name = "topicExchange", type = ExchangeTypes.TOPIC),
key = {"japan.#"}
))/*Topic话题*/
public void listenTopicQueue1(Object msg) {
System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
}
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "topic.queue3"),
exchange = @Exchange(name = "topicExchange", type = ExchangeTypes.TOPIC),
key = {"#.watch"}
))/*Topic话题*/
public void listenTopicQueue3(Object msg) {
System.out.println("消费者接收到topic.queue3的消息：【" + msg + "】");
}
}