RabbitMQ入门demo

落日映苍穹つ 2021-09-30 02:50 393阅读 0赞

引入依赖：

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

新建消息生产者：

package demo1;
    
    import com.rabbitmq.client.*;
    
    public class RabbitProducer {
    
        private static final String EXCHANGE_NAME = "exchange_demo";
        private static final String ROUTING_KEY = "routingkey_demo";
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADDRESS = "140.143.132.167";
        private static final int PORT = 5672;
    
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost(IP_ADDRESS);
            factory.setPort(PORT);
            factory.setUsername("chen");
            factory.setPassword("123");
            factory.setVirtualHost("/chen");
    
            Connection connection = factory.newConnection();
    
            Channel channel = connection.createChannel();
    
            channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
    
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
    
            String message = "hello world!";
    
            long start = System.currentTimeMillis();
    
            for (int i = 0; i < 10000000; i++) {
                channel.basicPublish(EXCHANGE_NAME,ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN,
                        message.getBytes());
            }
    
            long end = System.currentTimeMillis();
    
    
            System.out.println("共耗时：" + (end - start));
            channel.close();
            connection.close();
        }
    
    }

消息消费者：

package demo1;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    
    public class RabbitConsumer {
        private static final String EXCHANGE_NAME = "exchange_demo";
        private static final String ROUTING_KEY = "routingkey_demo";
        private static final String QUEUE_NAME = "queue_demo";
        private static final String IP_ADDRESS = "140.143.132.167";
        private static final int PORT = 5672;
    
    
        public static void main(String[] args) throws Exception {
            Address[] addresses = new Address[]{
                    new Address(IP_ADDRESS,PORT)
            };
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUsername("chen");
            factory.setPassword("123");
            factory.setVirtualHost("/chen");
    
            Connection connection = factory.newConnection(addresses);
    
    
            final Channel channel = connection.createChannel();
            channel.basicQos(64);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("recv msg:"+new String(body));
    
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (Exception e){
                        e.printStackTrace();
                    }
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }
            };
            channel.basicConsume(QUEUE_NAME,consumer);
    
            TimeUnit.SECONDS.sleep(5);
            channel.close();
            connection.close();
        }
    }