pom文件：
com.rabbitmq
amqp-client
5.0.0
org.springframework.amqp
spring-rabbit
2.0.2.RELEASE
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-amqp
连接工具类：
package top.wj.rabbitmq.client.utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class ChannelUtils {
public static Channel getChannelInstance(String connectionDescription) {
try {
ConnectionFactory connectionFactory = getConnectionFactory();
Connection connection = connectionFactory.newConnection(connectionDescription);
return connection.createChannel();
} catch (Exception e) {
throw new RuntimeException("获取Channel连接失败");
}
}private static ConnectionFactory getConnectionFactory() {
ConnectionFactory connectionFactory = new ConnectionFactory();
// 配置连接信息
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
// 网络异常自动连接恢复
connectionFactory.setAutomaticRecoveryEnabled(true);
// 每10秒尝试重试连接一次
connectionFactory.setNetworkRecoveryInterval(10000);
return connectionFactory;
}
}创建生产者：
package top.wj.rabbitmq.client.producer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import top.wj.rabbitmq.client.utils.ChannelUtils;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class MessageProducer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = ChannelUtils.getChannelInstance("队列消息生产者");
// 声明交换机 (交换机名, 交换机类型, 是否持久化, 是否自动删除, 是否是内部交换机, 交换机属性);
channel.exchangeDeclare("rabbitmq.wj", BuiltinExchangeType.DIRECT, true, false, false, new HashMap<>());
// 设置消息属性 发布消息 (交换机名, Routing key, 可靠消息相关属性 后续会介绍, 消息属性, 消息体);
AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().deliveryMode(2).contentType("UTF-8").build();
channel.basicPublish("rabbitmq.wj", "add", false, basicProperties, "body中的消息内容！".getBytes());
}
}创建消费者：
package top.wj.rabbitmq.client.consumer;
import com.rabbitmq.client.*;
import top.wj.rabbitmq.client.utils.ChannelUtils;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;
public class MessageConsumer {
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = ChannelUtils.getChannelInstance("队列消息消费者");
// 声明队列 (队列名, 是否持久化, 是否排他, 是否自动删除, 队列属性);
AMQP.Queue.DeclareOk declareOk = channel.queueDeclare("rabbitmq.wj.add", true, false, false, new HashMap<>());
// 声明交换机 (交换机名, 交换机类型, 是否持久化, 是否自动删除, 是否是内部交换机, 交换机属性);
channel.exchangeDeclare("rabbitmq.wj", BuiltinExchangeType.DIRECT, true, false, false, new HashMap<>());
// 将队列Binding到交换机上 (队列名, 交换机名, Routing key, 绑定属性);
channel.queueBind(declareOk.getQueue(), "rabbitmq.wj", "add", new HashMap<>());
// 消费者订阅消息 监听如上声明的队列 (队列名, 是否自动应答(与消息可靠有关 后续会介绍), 消费者标签, 消费者)
DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(consumerTag);
System.out.println(envelope.toString());
System.out.println(properties.toString());
System.out.println("消息内容:" + new String(body));
}
};
channel.basicConsume(declareOk.getQueue(), true, "消费者标签",defaultConsumer );
}
}控制台打印信息：
消费者标签
Envelope(deliveryTag=1, redeliver=false, exchange=rabbitmq.wj, routingKey=add)
#contentHeader(content-type=UTF-8, content-encoding=null, headers=null, delivery-mode=2, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
消息内容:body中的消息内容！ rabbitmq管理界面显示：
文章图片
文章图片
文章图片
【RabbitMQ的使用（五）RabbitMQ Java Client简单生产者、消费者代码示例】
文章图片
