8.RabbitMQ 消息传递Java对象
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了8.RabbitMQ 消息传递Java对象，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含4833字，纯文字阅读大概需要7分钟。
内容图文
通过消息服务器传递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();
}
}
}
原文：https://www.cnblogs.com/zzpblogs/p/8168808.html
内容总结
以上是互联网集市为您收集整理的8.RabbitMQ 消息传递Java对象全部内容，希望文章能够帮你解决8.RabbitMQ 消息传递Java对象所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】