RabbitMQ安装和测试小记

RabbitMQ安装和测试小记 收藏

操作系统CentOS x86_64 5.4

安装Erlang

可以到www.erlang.org去下载最新的源码，编译安装。

不过RabbitMQ的官方文档建议使用yum安装。

EPEL是啥？

参见http:///wiki/EPEL/FAQ#howtouse

安装步骤：

wget http://download.fedora./pub/epel/5/x86_64/epel-release-5-3.noarch.rpm

rpm -Uvh epel-release-5-3.noarch.rpm

yum install erlang

安装RabbitMQ

到 http://www./server.html 下载

rabbitmq-server-1.7.2-1.x86_64.rpm

然后

rpm -ivh rabbitmq-server-1.7.2-1.x86_64.rpm

运行RabbitMQ

/etc/init.d/rabbitmq-server start

注意

RabbitMQ使用了5672端口，修改iptables把5672端口打开。

测试

在官网下载java的客户端rabbitmq-java-client-bin-1.7.2.tar.gz

消息发布

 

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConnectionParameters;
import com.rabbitmq.client.MessageProperties;
public class PublishTest {
    public static void main(String[] args) {
        try {
            ConnectionParameters params = new ConnectionParameters();
            params.setUsername("guest");
            params.setPassword("guest");
            params.setVirtualHost("/");
            params.setRequestedHeartbeat(0);
            ConnectionFactory factory = new ConnectionFactory(params);
            Connection conn = factory.newConnection("192.168.56.110", 5672);
            Channel channel = conn.createChannel();
            channel.exchangeDeclare("exchange1", "direct", true);
            channel.queueDeclare("queue1", true);
            channel.queueBind("queue1", "exchange1", "key1");
            byte[] messageBodyBytes = "Hello, world!".getBytes();
            channel.basicPublish("exchange1", "key1", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
            channel.close();
            conn.close();
            System.out.println("Over.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.ConnectionParameters; import com.rabbitmq.client.MessageProperties; public class PublishTest { public static void main(String[] args) { try { ConnectionParameters params = new ConnectionParameters(); params.setUsername("guest"); params.setPassword("guest"); params.setVirtualHost("/"); params.setRequestedHeartbeat(0); ConnectionFactory factory = new ConnectionFactory(params); Connection conn = factory.newConnection("192.168.56.110", 5672); Channel channel = conn.createChannel(); channel.exchangeDeclare("exchange1", "direct", true); channel.queueDeclare("queue1", true); channel.queueBind("queue1", "exchange1", "key1"); byte[] messageBodyBytes = "Hello, world!".getBytes(); channel.basicPublish("exchange1", "key1", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes); channel.close(); conn.close(); System.out.println("Over."); } catch (Exception e) { e.printStackTrace(); } } } 

 

消息接收

 

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConnectionParameters;
import com.rabbitmq.client.GetResponse;
public class ConsumerTest {
    public static void main(String[] args) {
        try {
            ConnectionParameters params = new ConnectionParameters();
            params.setUsername("guest");
            params.setPassword("guest");
            params.setVirtualHost("/");
            params.setRequestedHeartbeat(0);
            ConnectionFactory factory = new ConnectionFactory(params);
            Connection conn = factory.newConnection("192.168.56.110", 5672);
            Channel channel = conn.createChannel();
            boolean noAck = false;
            GetResponse response = channel.basicGet("queue1", noAck);
            if (response == null) {
                System.out.println("No message!");
            } else {
                byte[] body = response.getBody();
                long deliveryTag = response.getEnvelope().getDeliveryTag();
                String str = new String(body);
                System.out.println(str);
                channel.basicAck(deliveryTag, false); // acknowledge receipt of the message
            }
            channel.close();
            conn.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.ConnectionParameters; import com.rabbitmq.client.GetResponse; public class ConsumerTest { public static void main(String[] args) { try { ConnectionParameters params = new ConnectionParameters(); params.setUsername("guest"); params.setPassword("guest"); params.setVirtualHost("/"); params.setRequestedHeartbeat(0); ConnectionFactory factory = new ConnectionFactory(params); Connection conn = factory.newConnection("192.168.56.110", 5672); Channel channel = conn.createChannel(); boolean noAck = false; GetResponse response = channel.basicGet("queue1", noAck); if (response == null) { System.out.println("No message!"); } else { byte[] body = response.getBody(); long deliveryTag = response.getEnvelope().getDeliveryTag(); String str = new String(body); System.out.println(str); channel.basicAck(deliveryTag, false); // acknowledge receipt of the message } channel.close(); conn.close(); } catch (Exception e) { e.printStackTrace(); } } }