8.RabbitMQ 消息传递Java对象

通过消息服务器传递Java对象，Java类必须实现序列化接口，可以把Java对象转化为字节数组，从消费者或生产者传递到另外一个JVM中，一定需要两个JVM共享这个类，比如是UserInfo类。

 

1、定义序列化的类UserInfo

2、消费者中，实例化UserInfo的对象，并取出它的字节数组

3、编写生产者

代码：

 

UserInfo.java

 

package com.test.rfc;

 

public class UserInfo implements java.io.Serializable{

private String name = null;

 

public String getName() {

return name;

}

 

public void setName(String name) {

this.name = name;

}

}

 

Server.java

 

package com.test.rfc;

 

import java.io.ByteArrayOutputStream;

import java.io.IOException;

import java.io.ObjectOutputStream;

 

import com.rabbitmq.client.AMQP;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.Consumer;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

 

public class Server {

public byte[] getUserByte() throws Exception

{

UserInfo u = new UserInfo();

u.setName("Hello I come from MQ server");

ByteArrayOutputStream baos = new ByteArrayOutputStream();

ObjectOutputStream oos = new ObjectOutputStream(baos);

oos.writeObject(u);

oos.close();

baos.close();

return baos.toByteArray();

}

public static void main(String[] argv) {

Server s = new Server();

ConnectionFactory factory = new ConnectionFactory();

    factory.setUsername("admin");

    factory.setPassword("admin");

        factory.setHost("192.168.169.142"); //使用默认端口5672

        Connection connection = null;

        

try {

connection = factory.newConnection();

final Channel channel = connection.createChannel();

//序列化对象

final byte[] data = s.getUserByte();

System.out.println(data.length);

String queueName = "queue_rpc";

channel.queueDeclare(queueName, false, false, false, null);

Consumer consumer = new DefaultConsumer(channel) {

 

@Override

public void handleDelivery(String consumerTag,

Envelope envelope, AMQP.BasicProperties properties,

byte[] body) throws IOException 

{

System.out.println("rfc=" + new String(body));

AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()

.correlationId(properties.getCorrelationId())

.build();

channel.basicPublish("", properties.getReplyTo(),

replyProps, data);

channel.basicAck(envelope.getDeliveryTag(), false);

}

};

channel.basicConsume(queueName, false, consumer);

} catch (Exception e) {

e.printStackTrace();

}

}

 

}

 

Client.java

 

package com.test.rfc;

 

import java.io.ByteArrayInputStream;

import java.io.IOException;

import java.io.ObjectInputStream;

import java.util.UUID;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

 

import com.rabbitmq.client.*;

 

public class Client {

public static void main(String[] argv) {

try

{

//发送消息的队列,Server在这个队列上接受消息

String queueName = "queue_rpc";

ConnectionFactory factory = new ConnectionFactory();

    factory.setUsername("admin");

    factory.setPassword("admin");

       factory.setHost("192.168.169.142"); //使用默认端口5672

       Connection connection = null;

connection = factory.newConnection();

Channel channel = connection.createChannel();

//生成临时的队列,Client在这队列上等待Server返回信息,Server向这个队列发消息

String replyQueueName = channel.queueDeclare().getQueue();

//生成唯一ID

final String corrId = UUID.randomUUID().toString();

AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()

.correlationId(corrId).replyTo(replyQueueName).build();

//客户端发送RFC请求

channel.basicPublish("", queueName, props, "GetUserInfo".getBytes());

//Server返回消息

final BlockingQueue response = new ArrayBlockingQueue(1);

channel.basicConsume(replyQueueName, true,

new DefaultConsumer(channel) {

@Override

public void handleDelivery(String consumerTag,

Envelope envelope, AMQP.BasicProperties properties,

byte[] body) throws IOException {

System.out.println(properties.getCorrelationId()+",body="+body.length);

if (properties.getCorrelationId().equals(corrId)) {

response.offer(body);

}

}

});

byte[] b = response.take();

System.out.println(b.length);

//反序列化对象

ByteArrayInputStream bais = new ByteArrayInputStream(b);

ObjectInputStream oii = new ObjectInputStream(bais);

UserInfo u = (UserInfo)oii.readObject();

System.out.println(u.getName());

channel.close();

connection.close();

}

catch(Exception e)

{

e.printStackTrace();

}

}

}