IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Java RabbitMQ案例如何Java对接

wen java案例 22

本文目录导读：

Java RabbitMQ案例如何Java对接

  1. 添加Maven依赖
  2. 连接工厂配置
  3. 生产者案例
  4. 消费者案例
  5. 工作队列模式（公平分发）
  6. 发布/订阅模式（Fanout交换机）
  7. 路由模式（Direct交换机）
  8. 主题模式（Topic交换机）
  9. 连接池（生产环境推荐）
  10. 生产环境最佳实践

我来为您详细解释Java对接RabbitMQ的完整案例。

添加Maven依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.18.0</version>
</dependency>

连接工厂配置

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class RabbitMQConfig {
    public static ConnectionFactory createConnectionFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost"); // RabbitMQ服务器地址
        factory.setPort(5672);        // 默认端口
        factory.setUsername("guest"); // 用户名
        factory.setPassword("guest"); // 密码
        factory.setVirtualHost("/");  // 虚拟主机
        return factory;
    }
}

生产者案例

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
public class Producer {
    private static final String QUEUE_NAME = "hello_queue";
    public static void main(String[] args) {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明队列（如果不存在则创建）
            // 参数：队列名称，是否持久化，是否独占，是否自动删除，其他参数
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            String message = "Hello RabbitMQ!";
            // 发送消息
            // 参数：交换机，路由键，消息属性，消息体
            channel.basicPublish("", QUEUE_NAME, 
                MessageProperties.PERSISTENT_TEXT_PLAIN, 
                message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

消费者案例

import com.rabbitmq.client.*;
public class Consumer {
    private static final String QUEUE_NAME = "hello_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        // 设置回调函数处理消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            // 手动确认消息
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        // 取消回调
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println(" [x] Consumer cancelled: " + consumerTag);
        };
        // 消费消息
        // 参数：队列名称，是否自动确认，投递回调，取消回调
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
    }
}

工作队列模式（公平分发）

生产者

public class WorkProducer {
    private static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            // 发送多个消息
            for (int i = 0; i < 10; i++) {
                String message = "Task " + i;
                channel.basicPublish("", QUEUE_NAME, 
                    MessageProperties.PERSISTENT_TEXT_PLAIN, 
                    message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + message + "'");
            }
        }
    }
}

消费者（公平分发）

public class WorkConsumer {
    private static final String QUEUE_NAME = "work_queue";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        // 设置预取计数，每次只处理一个消息
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
            try {
                // 模拟处理时间
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                // 手动确认
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

发布/订阅模式（Fanout交换机）

生产者

public class FanoutProducer {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明交换机（fanout类型）
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            String message = "Broadcast message to all queues";
            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费者

public class FanoutConsumer {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        // 创建临时队列
        String queueName = channel.queueDeclare().getQueue();
        // 绑定队列到交换机
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

路由模式（Direct交换机）

生产者

public class DirectProducer {
    private static final String EXCHANGE_NAME = "direct_exchange";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "direct");
            String[] routingKeys = {"info", "warning", "error"};
            for (String routingKey : routingKeys) {
                String message = "Message with routing key: " + routingKey;
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + message + "'");
            }
        }
    }
}

消费者

public class DirectConsumer {
    private static final String EXCHANGE_NAME = "direct_exchange";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        String queueName = channel.queueDeclare().getQueue();
        // 绑定多个路由键
        channel.queueBind(queueName, EXCHANGE_NAME, "error");
        channel.queueBind(queueName, EXCHANGE_NAME, "warning");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

主题模式（Topic交换机）

生产者

public class TopicProducer {
    private static final String EXCHANGE_NAME = "topic_exchange";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            // 发送不同主题的消息
            String[] routingKeys = {
                "quick.orange.rabbit",
                "lazy.orange.elephant",
                "quick.orange.fox",
                "lazy.brown.fox",
                "lazy.pink.rabbit",
                "quick.brown.fox"
            };
            for (String routingKey : routingKeys) {
                String message = "Message for " + routingKey;
                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + message + "'");
            }
        }
    }
}

消费者

public class TopicConsumer {
    private static final String EXCHANGE_NAME = "topic_exchange";
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = RabbitMQConfig.createConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");
        String queueName = channel.queueDeclare().getQueue();
        // 使用通配符绑定
        channel.queueBind(queueName, EXCHANGE_NAME, "*.orange.*");
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + 
                delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
    }
}

连接池（生产环境推荐）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.apache.commons.pool2.BasePooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericObjectPool;
public class RabbitMQPool {
    private final GenericObjectPool<Channel> channelPool;
    public RabbitMQPool() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection;
        try {
            connection = factory.newConnection();
        } catch (Exception e) {
            throw new RuntimeException("Failed to create connection", e);
        }
        Connection finalConnection = connection;
        channelPool = new GenericObjectPool<>(new BasePooledObjectFactory<Channel>() {
            @Override
            public Channel create() throws Exception {
                return finalConnection.createChannel();
            }
            @Override
            public PooledObject<Channel> wrap(Channel channel) {
                return new DefaultPooledObject<>(channel);
            }
        });
        // 配置连接池
        channelPool.setMaxTotal(10);
        channelPool.setMaxIdle(5);
        channelPool.setMinIdle(2);
    }
    public Channel borrowChannel() throws Exception {
        return channelPool.borrowObject();
    }
    public void returnChannel(Channel channel) {
        channelPool.returnObject(channel);
    }
    public void close() {
        channelPool.close();
    }
}

生产环境最佳实践

public class ProductionExample {
    // 1. 使用连接池管理连接
    private static RabbitMQPool pool = new RabbitMQPool();
    // 2. 添加重试机制
    public void sendMessageWithRetry(String message, int maxRetries) {
        int retries = 0;
        while (retries < maxRetries) {
            try (Channel channel = pool.borrowChannel()) {
                channel.confirmSelect(); // 启用发布确认
                channel.basicPublish("", "queue_name", 
                    MessageProperties.PERSISTENT_TEXT_PLAIN, 
                    message.getBytes());
                if (channel.waitForConfirms()) {
                    System.out.println("Message confirmed");
                    return;
                }
            } catch (Exception e) {
                retries++;
                if (retries >= maxRetries) {
                    throw new RuntimeException("Failed to send after " + maxRetries + " retries", e);
                }
                // 等待后重试
                try {
                    Thread.sleep(1000 * retries);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
    // 3. 异常处理
    public void handleMessage() {
        try (Connection connection = new ConnectionFactory().newConnection();
             Channel channel = connection.createChannel()) {
            channel.basicQos(1); // 每次处理一个消息
            DeliverCallback callback = (tag, delivery) -> {
                try {
                    String message = new String(delivery.getBody(), "UTF-8");
                    // 处理消息
                    processMessage(message);
                    // 手动确认
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                } catch (Exception e) {
                    // 处理失败，重新入队
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                }
            };
            channel.basicConsume("queue_name", false, callback, tag -> {});
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private void processMessage(String message) {
        // 业务处理逻辑
    }
}
  • 使用连接池管理RabbitMQ连接
  • 实现消息确认机制
  • 配置适当的预取计数
  • 添加重试和异常处理
  • 生产环境建议使用Spring AMQP

上一篇Java消息发送案例怎么编写

下一篇Java Redis发布订阅案例如何实操

相关文章

抱歉，评论功能暂时关闭!