java连接rabbitmq基础的发送和接收消息_java mq发送消息和接收消息

安装基础使用使用请看上一篇文章

一、基础的发送和接收消息

1.创建一个工具类获取一个rabbitmq的连接

public Connection getConnection() {
   // 创建连接工厂
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost"); // 设置RabbitMQ服务器的地址
    factory.setPort(5672); // 设置RabbitMQ服务器的端口号，默认为5672
    factory.setUsername("guest"); // 设置用户名，默认为guest
    factory.setPassword("guest"); // 设置密码，默认为guest
    try {
        Connection connection = factory.newConnection();
        return connection;
    } catch (Exception e) {
        return null;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.编写生产者发送消息

//获取连接
Connection connection = RabbitMqUtils.getConnection();
if (connection != null){
    //创建通道
    Channel channel = connection.createChannel();
    //创建一个队列 QUEUE_NAME自定义队列名称
    channel.queueDeclare(QUEUE_NAME,false,false,false,null);
    String message = "hello word";
    channel.basicPublish("",QUEUE_NAME,null,message.getBytes());

    //关闭通道和连接
    channel.close();
    connection.close();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

3.消费者接收消息

if (connection != null){
	Channel channel = connection.createChannel();
	channel.queueDeclare(QUEUE_NAME,false,false,false,null);
	DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
	    @Override
	    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
	        String message = new String(body);
	        System.out.println("消息内容：" + message);
	    }
	};
	
	// 监听队列，参数二：是否自动进行消息确认
	channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

上面消费者获取完消息也不会停止程序，会一直监听看是否有新的消息，一点进入队列就会被打印；
4.消息确认机制（ACK）
上述方法为自动确认，一旦接受消费者自动发送ACK
手动确认ACK：消息接收后不会发送ACK，需要手动调用

if (connection != null){
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body);
        //逻辑处理一旦出问题就不会被消费无法ACK，消息还在队列内
        System.out.println("消息内容：" + message);

        //手动进行ACK
        channel.basicAck(envelope.getDeliveryTag(),false);
    }
};

// 监听队列，参数二：是否自动进行消息确认
channel.basicConsume(QUEUE_NAME,false,defaultConsumer);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18