使用Java原生的方式使用RabbitMQ现在已经较少，但这是基础，还是有必要了解的。
-
引入依赖。
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.16.0</version> </dependency> -
实现生产者。
// 使用ConnectionFactory创建连接 ConnectionFactory factory = new ConnectionFactory(); // 设置连接信息 factory.setHost("127.0.0.1"); // 注意这里写5672，是amqp协议端口 factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); factory.setVirtualHost("/test"); // 创建Connection Connection connection = factory.newConnection(); // 通过Connection创建新的Channel Channel channel = connection.createChannel(); try (connection; channel) { // 声明队列，如果此队列不存在，会自动创建 channel.queueDeclare("test_java", false, false, false, null); // 将队列绑定到交换机 channel.queueBind("test_java", "amq.direct", "test_java_key"); // 转换为byte[]，发布消息 channel.basicPublish("amq.direct", "test_java_key", null, "Hello World!".getBytes()); }-
queueDeclare方法参数如下：-
queue：队列的名称（默认创建后routingKey和队列名称一致）
-
durable：是否持久化。
-
exclusive：是否排他。
如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。
排他队列是基于Connection可见，同一个Connection的不同Channel是可以同时访问同一个连接创建的排他队列。
如果一个Connection已经声明了一个排他队列，其他的Connection是不允许建立同名的排他队列的。
即使该队列是持久化的，一旦Connection关闭或者客户端退出，该排他队列都会自动被删除。
-
autoDelete：是否自动删除。
-
arguments：设置队列的其他一些参数，这里暂时不需要。
-
-
queueBind方法参数如下：- queue：需要绑定的队列名称。
- exchange：需要绑定的交换机名称。
- routingKey：路由标识。
-
basicPublish方法参数如下：- exchange：对应的Exchange名称，这里就使用第二个直连交换机。
- routingKey：这里填写绑定时指定的routingKey，其实和之前在管理页面操作一样。
- props：其他的配置。
- body：消息本体。
执行完成后，可以在管理页面中看到刚刚创建好的消息队列：
并且此消息队列已经成功与
amq.direct交换机进行绑定：
-
-
实现消费者。
// 使用ConnectionFactory创建连接 ConnectionFactory factory = new ConnectionFactory(); // 设置连接信息 factory.setHost("127.0.0.1"); // 注意这里写5672，是amqp协议端口 factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); factory.setVirtualHost("/test"); // 这里不使用try-with-resource，因为消费者是一直等待新的消息到来，然后按照设定的逻辑进行处理，所以这里不能在定义完成之后就关闭连接 // 创建Connection Connection connection = factory.newConnection(); // 通过Connection创建新的Channel Channel channel = connection.createChannel(); // 创建一个基本的消费者 channel.basicConsume("test_java", false, (s, delivery) -> { System.out.println(new String(delivery.getBody())); /* 确认应答 参数 1：当前的消息标签 参数 2：是否批量处理消息队列中所有的消息，如果为false表示只处理当前消息 */ channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); /* 拒绝应答 参数 1：当前的消息标签 参数 2：倍数 参数 3：是否将当前消息放回队列，如果为false，那么消息就会被丢弃 */ // channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false); /* 拒绝应答 参数 1：当前的消息标签 参数 2：是否将当前消息放回队列，如果为false，那么消息就会被丢弃 */ // channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false); }, s -> { });basicConsume方法参数如下：-
queue：消息队列名称，直接指定。
-
autoAck：自动应答。
消费者从消息队列取出数据后，需要跟服务器进行确认应答，当服务器收到确认后，会自动将消息删除，如果开启自动应答，那么消息发出后会直接删除。
-
deliver：消息接收后的函数回调。
可以在回调中对消息进行处理，处理完成后，需要给服务器确认应答。
-
cancel：当消费者取消订阅时进行的函数回调，这里暂时用不到。
执行上述代码后，控制台输出：Hello World!，说明消息消费成功。
-
- 环境
- JDK 17.0.6
- Maven 3.6.3
- amqp-client 5.16.0