当前位置：首页 > news >正文

RabbitMQ-hello

news 2025/8/6 6:00:49

0. pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.hong.rabbitmq</groupId><artifactId>rabbitmq-hello</artifactId><version>1.0-SNAPSHOT</version><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.8.0</version><configuration><encoding>utf-8</encoding><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build><dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.8.0</version></dependency><dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId><version>2.6</version></dependency></dependencies>
</project>

1. 生产者代码

package com.hong.rabbitmq1;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {//队列名称private static final String QUEUE_NAME = "hello";public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();factory.setHost("10.211.55.4");factory.setUsername("admin");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();/*** 声明消息队列* 第一个参数: 队列名称* 第二个参数: 是否持久化     false不持久化,rabbitmq重启队列丢失  true持久化,rabbitmq重启队列不丢失* 第三个参数: 是否排外       false排外只能一个消费者消费         false不排外可多消费者消费* 第四个参数: 是否自动删除   true自动删除最后一个消费者断开连接后该队列自动删除   false不自动删除* 第五个参数: 其他参数信息*/channel.queueDeclare(QUEUE_NAME,false,false,false,null);String message = "HELLO RABBITMQ!";/*** 发布消息* 第一个参数:要将消息发送到哪个交换机* 第二个参数:路由的 key 是哪个* 第三个参数:其他参数信息* 第四个参数:消息的消息体*/channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("消息发送完成!");}}

在这里插入图片描述

2.消费者代码

package com.hong.rabbitmq1;import com.rabbitmq.client.*;public class Consumer {private static final String QUEUE_NAME = "hello";public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();factory.setHost("10.211.55.4");factory.setUsername("admin");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();DeliverCallback deliverCallback = (comsumerTag,message) -> System.out.println(new String(message.getBody()));CancelCallback cancelCallback = var -> System.out.println(var + "消息消费被中断!");/*** 消息消费* 第一个参数: 消费哪个队列* 第二个参数: 是否自动确认消息* 第三个参数: 当一个消息发送过来后的回调接口* 第四个参数: 当一个消费者取消订阅时的回调接口;取消消费者订阅队列时除了使用{@link Channel#basicCancel}之外的所有方式都会调用该回调方法*/channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);}
}

在这里插入图片描述

3.抽取工具类

package com.hong.utils;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** RabbitMQ创建Channel工具类*/
public class RabbitMQUtil {public static Channel getChannel() throws Exception{ConnectionFactory factory = new ConnectionFactory();factory.setHost("10.211.55.4");factory.setUsername("admin");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();return channel;}
}

4.工作队列

4.1.生产者代码

package com.hong.rabbitmq2;import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;import java.util.Scanner;public class Task {//队列名称private static final String QUEUE_NAME = "hello";public static void main(String[] args) throws Exception{Channel channel = RabbitMQUtil.getChannel();channel.queueDeclare(QUEUE_NAME,false,false,false,null);Scanner scanner = new Scanner(System.in);System.out.println("请输入:");while (scanner.hasNext()){String message = scanner.next();channel.basicPublish("",QUEUE_NAME,null,message.getBytes());System.out.println("消息发送完成------" + message);}}
}

4.2.消费者代码

package com.hong.rabbitmq2;import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Worker {private static final String QUEUE_NAME = "hello";public static void main(String[] args) throws Exception{Channel channel = RabbitMQUtil.getChannel();DeliverCallback deliverCallback = (comsumerTag, message) -> System.out.println("接收到的消息:"+  new String(message.getBody()));CancelCallback cancelCallback = var -> System.out.println(var + "消息消费被中断!");System.out.println("worker1等待接收消息");channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);}
}