1. 程式人生 > >RabbitMQ生產者和消費者程式碼

RabbitMQ生產者和消費者程式碼

生產者

package com.rabbitMQ.pro;  

import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.Channel;  

public class Producer {  

    private final static String QUEUE_NAME = "hello2";// 佇列名不能重複 之前已有就會失敗  

    public static void main(String[] argv) throws java.io
.IOException { /* 使用工廠類建立Connection和Channel,並且設定引數 */ ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.10.111");// MQ的IP factory.setPort(5672);// MQ埠 factory.setUsername("asdf");// MQ使用者名稱 factory.setPassword("123456");// MQ密碼
Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); /* 建立訊息佇列,並且傳送訊息 */ channel.queueDeclare(QUEUE_NAME, false, false, false, null); String message = "訊息2"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes
()); System.out.println("生產了個'" + message + "'"); /* 關閉連線 */ channel.close(); connection.close(); } }

消費者

package com.rabbitMQ.pro;  

import com.rabbitmq.client.ConnectionFactory;  
import com.rabbitmq.client.Connection;  
import com.rabbitmq.client.Channel;  
import com.rabbitmq.client.QueueingConsumer;  

public class Consumer {  

    private final static String QUEUE_NAME = "hello2";  

    public static void main(String[] argv) throws java.io.IOException,  
            java.lang.InterruptedException {  
        /* 建立連線 */  
        ConnectionFactory factory = new ConnectionFactory();  
        factory.setHost("192.168.10.111");// MQ的IP  
        factory.setPort(5672);// MQ埠  
        factory.setUsername("asdf");// MQ使用者名稱  
        factory.setPassword("123456");// MQ密碼  
        Connection connection = factory.newConnection();  
        Channel channel = connection.createChannel();  

        /* 宣告要連線的佇列 */  
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
        System.out.println("等待訊息產生:");  

        /* 建立消費者物件,用於讀取訊息 */  
        QueueingConsumer consumer = new QueueingConsumer(channel);  
        channel.basicConsume(QUEUE_NAME, true, consumer);  

        /* 讀取佇列,並且阻塞,即在讀到訊息之前在這裡阻塞,直到等到訊息,完成訊息的閱讀後,繼續阻塞迴圈 */  
        while (true) {  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
            String message = new String(delivery.getBody());  
            System.out.println("收到訊息'" + message + "'");  
        }  
    }  
}