rabbitmq，生成者和消費者

生成者

package cn.go4mi;  
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
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("localhost");// MQ的IP  
        factory.setPort(5672);// MQ端口  
        factory.setUsername("guest");// MQ用戶名  
        factory.setPassword("guest");// MQ密碼  
        Connection connection = factory.newConnection();  
        Channel channel = connection.createChannel();  
        /* 創建消息隊列，並且發送消息 */  
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);  
        for (int i = 0; i < 10; i++) {
            String message = "生成消息"+i;  
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());  
            System.out.println("生產了個'" + message + "'");  
        }
        /* 關閉連接 */  
        channel.close();  
        connection.close();  
    }  
}  

 

消費者

package cn.go4mi;  
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
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("localhost");// MQ的IP  
        factory.setPort(5672);// MQ端口  
        factory.setUsername("guest");// MQ用戶名  
        factory.setPassword("guest");// 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);  
        int i=1;
        /* 讀取隊列，並且阻塞，即在讀到消息之前在這里阻塞，直到等到消息，完成消息的閱讀后，繼續阻塞循環 */  
        while (true) {  
            QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
            String message = new String(delivery.getBody());  
            System.out.println("第"+i+"個消息！");
            System.out.println("收到消息'" + message + "'");  
            i++;
            Thread.sleep(1000*5);
        }  
    }  
}