当前位置：首页 > article >正文

Java项目--仿RabbitMQ的消息队列--基于MQ的生产者消费者模型

article 2024/12/24 5:59:22

一、引言

二、生产者

三、消费者

四、扩展

五、总结

一、引言

本篇文章就是本次Java项目的最后一篇文章了，本篇文章主要介绍基于MQ的生产者消费者模型的代码编写

二、生产者

public class DemoConsumer {
    public static void main(String[] args) throws IOException, InterruptedException, MqException {
        System.out.println("启动消费者！");
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(9090);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);
        channel.queueDeclare("testQueue",true,false,false,null);

        channel.basicConsume("testQueue", true, new Consumer() {
            @Override
            public void handleDelivery(String consumeTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                System.out.println("[消费数据]开始！");
                System.out.println("consumerTag="+consumeTag);
                System.out.println("basicProperties="+basicProperties);
                String bodyString = new String(body,0, body.length);
                System.out.println("body="+bodyString);
                System.out.println("[消费数据]结束！");
            }
        });

        while (true){
            Thread.sleep(500);
        }
    }
}

三、消费者

public class DemoProducer {
    public static void main(String[] args) throws IOException, InterruptedException {
        System.out.println("启动生产者！");
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("127.0.0.1");
        factory.setPort(9090);
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);
        channel.queueDeclare("testQueue",true,false,false,null);

        byte[] body = "hello".getBytes();
        boolean ok = channel.basicPublish("testExchange","testQueue",null,body);
        System.out.println("消息投递完成！ok="+ok);
        Thread.sleep(500);

        channel.close();
        connection.close();
    }
}