1 简单队列

目录

• 1、获取MQ连接
• 2、发送消息
• 3、消费者旧API
• 4、消费者新API

采用老的API实现的，所谓的简单队列就是一个消费者一个生产者，是1：1的关系

• P ：消息的生产者
• 红色的：队列
• C：消费者

1、获取MQ连接

public class MqConnectionUtil {
    private static final Logger log =LoggerFactory.getLogger(RedisUtil.class);
    public static Connection getConnection() {
        Connection connection = null;
        ConnectionFactory factory = new ConnectionFactory();
        // 用户名
        factory.setUsername("guest");
        // 密码
        factory.setPassword("guest'");
        // 服务器地址
        factory.setHost("47.*.*.9");
        // 端口号，也就是AMQP
        factory.setPort(5672);
        // 数据库 “/”代表所有的数据库
        factory.setVirtualHost("/");
        try {
            connection = factory.newConnection();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            log.error("RabbitMQ connection create failed!!");
            e.printStackTrace();
        }
        return connection;
    }
}

2、发送消息

public static void main(String[] args) throws IOException, TimeoutException {

        // 获取连接
        Connection connection = MqConnectionUtil.getConnection();

        // 获取通道
        Channel channel = connection.createChannel();

        // 创建队列申明
        channel.queueDeclare(QUENU_NAME,false,false,false,null);


        channel.basicPublish("",QUENU_NAME,null,"hello zhaodi".getBytes());
        channel.close();
        connection.close();
    }

3、消费者旧API

这段代码中的许多方法已经被JAVA弃用，但是目前我们作为学习的使用

public static void main(String[] args) throws IOException, InterruptedException {

        Connection connection = MqConnectionUtil.getConnection();


        Channel channel = connection.createChannel();

        // 定义消费者
        QueueingConsumer consumer = new QueueingConsumer(channel);

        //监听队列
        channel.basicConsume(QUENU_NAME,true,consumer);
        while(true) {

            Delivery delivery = consumer.nextDelivery();

            System.out.println("收到的消息 ："+new String(delivery.getBody()));
        }


    }

4、消费者新API

public class Consumer2 {

    private static final String QUEUE_NAME = "my-simple-queue";
    public static void main(String[] args) throws IOException, InterruptedException {

        Connection connection = MqConnectionUtil.getConnection();

        Channel channel = connection.createChannel();
        // 申明队列  如果该队列已经存在/或者生产者已经申明了在这里就不需要再次申明
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 定义消费者
        DefaultConsumer consumer = new DefaultConsumer(channel){

            /**
             *获取到到达的消息，触发回调事件
             *
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body);
                System.out.println("接受的消息是："+msg);
            }
        };

        //监听队列
        channel.basicConsume(QUEUE_NAME,true,consumer);
    }
}