生产者
package org.example.helloworld;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.time.LocalDateTime;
import java.util.concurrent.TimeoutException;
public class Send {
private final static String QUEUE_NAME = "hello";
public static void main(String[] args) throws IOException, TimeoutException {
//TCP 物理连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.121.200");
try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
// 宣布队列
// 形参:
// queue – 队列的名称
// durable – 如果我们声明一个持久队列,则为true(该队列将在服务器重启后继续存在)
// exclusive – 如果我们声明的是独占队列(仅限于此连接),则为true
// autoDelete – 如果我们声明一个自动删除队列(当不再使用时,服务器将删除它),则为true
// arguments – 队列的其他属性(构造参数)
// 返回值:
// a declaration-方法来指示队列已成功声明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!" + LocalDateTime.now();
// 发布消息。发布到不存在的交换将导致通道级协议异常,从而关闭通道。
// 如果资源驱动的警报生效,通道#basicPublish的调用将最终阻塞。
// 形参:
// exchange – 交换机名称
// routingKey – 队列名称
// props – 其他属性
// message - 路由头
// body – 消息正文
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
}
}
}
消费者
package org.example.helloworld;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
public class Recv {
private final static String QUEUE_NAME = "hello";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.121.200");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 宣布队列
// 形参:
// queue – 队列的名称
// durable – 如果我们声明一个持久队列,则为true(该队列将在服务器重启后继续存在)
// exclusive – 如果我们声明的是独占队列(仅限于此连接),则为true
// autoDelete – 如果我们声明一个自动删除队列(当不再使用时,服务器将删除它),则为true
// arguments – 队列的其他属性(构造参数)
// 返回值:
// a declaration-方法来指示队列已成功声明
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
// 使用服务器生成的consumerTag启动非本地、非独占的消费者。
// 仅提供对basic.deliver和basic.cancel AMQP方法的访问(这对于大多数情况来说已经足够了)。
// 查看具有消费者参数的方法,以访问所有应用程序回调。
//形参:
// queue – 队列名称
// autoAck – 如果服务器认为消息一经传递即被确认,则为true,如果服务器需要显式确认,则为false
// callback – 消费者对象的接口
//返回值:
// 服务器生成的消费者标签
channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, StandardCharsets.UTF_8);
System.out.println(" [x] Received '" + message + "'");
//签收消息,确认消息
//envelope.getDeliveryTag() 获取这个消息的TagId
//false只确认签收当前的消息,设置为true的时候则代表签收该消费者所有未签收的消息
channel.basicAck(envelope.getDeliveryTag() , false);
}
});
}
}
标签:false,String,方式,队列,HelloWorld,简单,import,true,channel
From: https://www.cnblogs.com/xl4ng/p/16908057.html