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();
}
}
}
