RabbitMQ的使用（五）RabbitMQ Java Client简单生产者、消费者代码示例

青旅半醒 2022-05-13 22:56 165阅读 0赞

# pom文件： #

<dependencies>
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.0.0</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.amqp</groupId>
                <artifactId>spring-rabbit</artifactId>
                <version>2.0.2.RELEASE</version>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
    
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
        </dependencies>

# 连接工具类： #

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<basic>(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管理界面显示： #

![70][]![70 1][]

![70 2][]

![70 3][]

[70]: https://img-blog.csdn.net/20180830155333233?watermark/2/text/aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3c4OTM5MzI3NDc=/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70
[70 1]: /images/20220514/cfa623594ba14e979904a384b6c0558a.png
[70 2]: /images/20220514/4bfec93749f94e3fb13883f0f160944d.png
[70 3]: /images/20220514/904b433c96774df0ac6802fdf8a6275f.png