rabbitmq 学习-2-初试
本例是一个简单的异步发送消息实例
1，发送端
@Test(groups = { "sunjun" })
public class RabbitmqTest {

private static Connection connection;

static {
ConnectionParameters params = new ConnectionParameters();
ConnectionFactory factory = new ConnectionFactory(params);
try {
connection = factory.newConnection("192.168.18.21",
AMQP.PROTOCOL.PORT);
} catch (IOException e) {
e.printStackTrace();
}

}

public void testSend() {
try {
Channel channel = connection.createChannel();
System.out.println(channel.toString());
Assert.assertNotNull(channel);
byte[] messageBodyBytes = "hello world".getBytes();
channel.basicPublish("exchangeName", "routingKey",
MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);
} catch (Exception e) {
e.printStackTrace();
}
}
}

2，接收端
@Test(groups = { "sunjun" })
public class RabbitmqTestB {

private static Connection connection;

static {
ConnectionParameters params = new ConnectionParameters();
ConnectionFactory factory = new ConnectionFactory(params);
try {
connection = factory.newConnection("localhost", AMQP.PROTOCOL.PORT);
} catch (IOException e) {
e.printStackTrace();
}

}

public void testReceive() {
try {
Channel channel = connection.createChannel();
System.out.println(channel.toString());

Assert.assertNotNull(channel);

channel.exchangeDeclare("exchangeName", "direct");
channel.queueDeclare("queueName");
channel.queueBind("queueName", "exchangeName", "routingKey");

boolean noAck = false;
GetResponse response = channel.basicGet("queueName", true);
if (response == null) {
System.out.println("No message retrieved.");
} else {
AMQP.BasicProperties props = response.getProps();
byte[] body = response.getBody();
System.out.println(new String(body));
}
} catch (IOException e) {
e.printStackTrace();
}
}
}

先执行下发送端，再执行下接收端，输出：hello world
ok

----------------------------------
RpcClient，RpcServer同步发送接收消息
Channel.basicPublish，Channel.basicGet异步发送接收消息
本例是一个简单的同步发送消息实例
1，发送端
public class Publish {

private static Connection connection;

static {
ConnectionParameters params = new ConnectionParameters();
ConnectionFactory factory = new ConnectionFactory(params);
try {
connection = factory.newConnection("localhost", AMQP.PROTOCOL.PORT);
} catch (IOException e) {
e.printStackTrace();
}
}

public static void main(String[] args) {
try {
Channel channel = connection.createChannel();

RpcClient rpc = new RpcClient(channel, "exchangeName", "routingKey");
byte[] primitiveCall = rpc.primitiveCall("hello world".getBytes());
System.out.println(new String(primitiveCall));
primitiveCall = rpc.primitiveCall("hello world2".getBytes());
System.out.println(new String(primitiveCall));

rpc = new RpcClient(channel, "exchangeName", "routingKey2");
primitiveCall = rpc.primitiveCall("hello world2".getBytes());
System.out.println(new String(primitiveCall));

System.out.println("publish success.");
} catch (Exception e) {
e.printStackTrace();
}
}
}

2，接收端
public class Receive {

private static Connection connection;

static {
ConnectionParameters params = new ConnectionParameters();
ConnectionFactory factory = new ConnectionFactory(params);
try {
connection = factory.newConnection("localhost", AMQP.PROTOCOL.PORT);
} catch (IOException e) {
e.printStackTrace();
}
}

public static void main(String[] args) {
try {
Channel channel = connection.createChannel();
System.out.println(channel.toString());

channel.exchangeDeclare("exchangeName", "topic");
channel.exchangeDeclare("exchangeName2", "topic");
channel.queueDeclare("queueName");
channel.queueBind("queueName", "exchangeName", "routingKey");
channel.queueBind("queueName", "exchangeName", "routingKey2");
channel.queueBind("queueName", "exchangeName2", "routingKey2");
channel.queueBind("queueName", "exchangeName2", "routingKey");
//queue 与 exchange 是多对多的，可以把同一queue和exchange以多个不同的routing进行bind，这样就会有多个routing，而不是一个，虽然说这些rout 是绑定相同的 exchange, queue

final RpcServer rpcServer = new RpcServer(channel, "queueName") {
@Override
public byte[] handleCall(byte[] requestBody, AMQP.BasicProperties replyProperties) {
System.out.println("receive msg: " + new String(requestBody));
return "return message".getBytes();
}
};

Runnable main = new Runnable() {
@Override
public void run() {
try {
throw rpcServer.mainloop();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
};
new Thread(main).start();

System.out.println("receive success.");
} catch (IOException e) {
e.printStackTrace();
}
}
}