PHP小码农

Appearance

Java 消息序列化 ​

一、概述 ​

消息序列化是将对象转换为可传输格式的过程，反序列化则是将接收到的数据还原为对象。在 RabbitMQ 中，合理的序列化策略对于系统的性能、兼容性和安全性都至关重要。

1.1 序列化流程 ​

┌─────────────────────────────────────────────────────────────────────┐
│                       序列化流程                                     │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│  生产者端：                                                         │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐  │
│  │  Object  │ ──► │Serialize │ ──► │  byte[]  │ ──► │ RabbitMQ │  │
│  │  (对象)   │     │ (序列化)  │     │ (字节数组) │     │  Broker  │  │
│  └──────────┘     └──────────┘     └──────────┘     └──────────┘  │
│                                                                     │
│  消费者端：                                                         │
│  ┌──────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐  │
│  │ RabbitMQ │ ──► │  byte[]  │ ──► │Deserialize│ ──► │  Object  │  │
│  │  Broker  │     │ (字节数组) │     │ (反序列化) │     │  (对象)   │  │
│  └──────────┘     └──────────┘     └──────────┘     └──────────┘  │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

1.2 常见序列化格式 ​

text
┌─────────────────────────────────────────────────────────────────────┐
│                     序列化格式对比                                   │
├───────────────┬─────────────┬─────────────┬─────────────────────────┤
│     格式      │   可读性    │   性能      │         特点            │
├───────────────┼─────────────┼─────────────┼─────────────────────────┤
│     JSON      │    高       │    中       │ 通用、跨语言、易调试    │
├───────────────┼─────────────┼─────────────┼─────────────────────────┤
│   MessagePack │    中       │    高       │ 二进制、体积小、跨语言  │
├───────────────┼─────────────┼─────────────┼─────────────────────────┤
│  Protobuf     │    低       │    很高     │ 高效、需要定义 Schema   │
├───────────────┼─────────────┼─────────────┼─────────────────────────┤
│   Java 序列化 │    无       │    低       │ Java 专用、不推荐       │
├───────────────┼─────────────┼─────────────┼─────────────────────────┤
│     XML       │    高       │    低       │ 冗长、可读性好          │
└───────────────┴─────────────┴─────────────┴─────────────────────────┘

二、核心知识点 ​

2.1 JSON 序列化 ​

2.1.1 使用 Jackson ​

java
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.rabbitmq.client.*;

public class JsonSerialization {
    
    private final ObjectMapper objectMapper;
    
    public JsonSerialization() {
        objectMapper = new ObjectMapper();
        objectMapper.registerModule(new JavaTimeModule());
        objectMapper.disable(com.fasterxml.jackson.databind.SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
    }
    
    public byte[] serialize(Object obj) throws Exception {
        return objectMapper.writeValueAsBytes(obj);
    }
    
    public <T> T deserialize(byte[] data, Class<T> clazz) throws Exception {
        return objectMapper.readValue(data, clazz);
    }
    
    public void publish(Channel channel, String exchange, String routingKey, 
                       Object message) throws Exception {
        
        byte[] body = serialize(message);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .contentEncoding("UTF-8")
            .build();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
    
    public <T> T consume(Delivery delivery, Class<T> clazz) throws Exception {
        return deserialize(delivery.getBody(), clazz);
    }
}

2.1.2 消息对象定义 ​

java
import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.Map;

public class OrderMessage {
    
    private String orderId;
    private String userId;
    private double amount;
    private String status;
    private LocalDateTime createdAt;
    private Map<String, Object> metadata;
    
    public OrderMessage() {}
    
    public OrderMessage(String orderId, String userId, double amount) {
        this.orderId = orderId;
        this.userId = userId;
        this.amount = amount;
        this.status = "CREATED";
        this.createdAt = LocalDateTime.now();
    }
    
    public String getOrderId() { return orderId; }
    public void setOrderId(String orderId) { this.orderId = orderId; }
    
    public String getUserId() { return userId; }
    public void setUserId(String userId) { this.userId = userId; }
    
    public double getAmount() { return amount; }
    public void setAmount(double amount) { this.amount = amount; }
    
    public String getStatus() { return status; }
    public void setStatus(String status) { this.status = status; }
    
    public LocalDateTime getCreatedAt() { return createdAt; }
    public void setCreatedAt(LocalDateTime createdAt) { this.createdAt = createdAt; }
    
    public Map<String, Object> getMetadata() { return metadata; }
    public void setMetadata(Map<String, Object> metadata) { this.metadata = metadata; }
    
    @Override
    public String toString() {
        return "OrderMessage{" +
            "orderId='" + orderId + '\'' +
            ", userId='" + userId + '\'' +
            ", amount=" + amount +
            ", status='" + status + '\'' +
            ", createdAt=" + createdAt +
            '}';
    }
}

2.1.3 JSON 消息处理 ​

java
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.*;

public class JsonMessageHandler {
    
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    public void publishOrder(Channel channel, String exchange, String routingKey,
                            OrderMessage order) throws Exception {
        
        String json = objectMapper.writeValueAsString(order);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .contentEncoding("UTF-8")
            .messageId(order.getOrderId())
            .type("order.created")
            .timestamp(new java.util.Date())
            .build();
        
        channel.basicPublish(exchange, routingKey, props, json.getBytes("UTF-8"));
        
        System.out.println("订单消息已发布: " + order.getOrderId());
    }
    
    public OrderMessage consumeOrder(Delivery delivery) throws Exception {
        String contentType = delivery.getProperties().getContentType();
        
        if (!"application/json".equals(contentType)) {
            throw new IllegalArgumentException("不支持的内容类型: " + contentType);
        }
        
        String json = new String(delivery.getBody(), "UTF-8");
        
        return objectMapper.readValue(json, OrderMessage.class);
    }
}

2.2 MessagePack 序列化 ​

java
import org.msgpack.core.MessageBufferPacker;
import org.msgpack.core.MessagePack;
import org.msgpack.core.MessageUnpacker;
import com.rabbitmq.client.*;

import java.util.Map;

public class MessagePackSerialization {
    
    public byte[] serializeOrder(OrderMessage order) throws Exception {
        MessageBufferPacker packer = MessagePack.newDefaultBufferPacker();
        
        packer.packMapHeader(6)
            .packString("orderId").packString(order.getOrderId())
            .packString("userId").packString(order.getUserId())
            .packString("amount").packDouble(order.getAmount())
            .packString("status").packString(order.getStatus())
            .packString("createdAt").packString(order.getCreatedAt().toString())
            .packString("metadata").packMapHeader(0);
        
        packer.close();
        return packer.toByteArray();
    }
    
    public OrderMessage deserializeOrder(byte[] data) throws Exception {
        MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(data);
        
        OrderMessage order = new OrderMessage();
        
        int mapSize = unpacker.unpackMapHeader();
        for (int i = 0; i < mapSize; i++) {
            String key = unpacker.unpackString();
            switch (key) {
                case "orderId":
                    order.setOrderId(unpacker.unpackString());
                    break;
                case "userId":
                    order.setUserId(unpacker.unpackString());
                    break;
                case "amount":
                    order.setAmount(unpacker.unpackDouble());
                    break;
                case "status":
                    order.setStatus(unpacker.unpackString());
                    break;
                case "createdAt":
                    order.setCreatedAt(LocalDateTime.parse(unpacker.unpackString()));
                    break;
                default:
                    unpacker.skipValue();
            }
        }
        
        unpacker.close();
        return order;
    }
    
    public void publishWithMessagePack(Channel channel, String exchange, 
                                       String routingKey, OrderMessage order) 
            throws Exception {
        
        byte[] body = serializeOrder(order);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/msgpack")
            .messageId(order.getOrderId())
            .build();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
}

2.3 Protobuf 序列化 ​

2.3.1 Protobuf 定义 ​

protobuf
syntax = "proto3";

package com.example.rabbitmq;

option java_package = "com.example.rabbitmq.proto";
option java_outer_classname = "OrderProto";

message Order {
    string order_id = 1;
    string user_id = 2;
    double amount = 3;
    string status = 4;
    int64 created_at = 5;
    map<string, string> metadata = 6;
}

message OrderList {
    repeated Order orders = 1;
}

2.3.2 Protobuf 序列化实现 ​

java
import com.example.rabbitmq.proto.OrderProto;
import com.rabbitmq.client.*;

public class ProtobufSerialization {
    
    public byte[] serialize(OrderProto.Order order) {
        return order.toByteArray();
    }
    
    public OrderProto.Order deserialize(byte[] data) throws Exception {
        return OrderProto.Order.parseFrom(data);
    }
    
    public void publish(Channel channel, String exchange, String routingKey,
                       OrderProto.Order order) throws Exception {
        
        byte[] body = order.toByteArray();
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/x-protobuf")
            .messageId(order.getOrderId())
            .type("order.created")
            .build();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
    
    public OrderProto.Order consume(Delivery delivery) throws Exception {
        return OrderProto.Order.parseFrom(delivery.getBody());
    }
    
    public OrderProto.Order createOrder(String orderId, String userId, 
                                        double amount) {
        return OrderProto.Order.newBuilder()
            .setOrderId(orderId)
            .setUserId(userId)
            .setAmount(amount)
            .setStatus("CREATED")
            .setCreatedAt(System.currentTimeMillis())
            .build();
    }
}

2.4 自定义序列化器 ​

java
import com.rabbitmq.client.*;

public interface MessageSerializer {
    
    byte[] serialize(Object obj) throws Exception;
    
    <T> T deserialize(byte[] data, Class<T> clazz) throws Exception;
    
    String getContentType();
}

public class JsonMessageSerializer implements MessageSerializer {
    
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    @Override
    public byte[] serialize(Object obj) throws Exception {
        return objectMapper.writeValueAsBytes(obj);
    }
    
    @Override
    public <T> T deserialize(byte[] data, Class<T> clazz) throws Exception {
        return objectMapper.readValue(data, clazz);
    }
    
    @Override
    public String getContentType() {
        return "application/json";
    }
}

public class MessagePackSerializer implements MessageSerializer {
    
    @Override
    public byte[] serialize(Object obj) throws Exception {
        return null;
    }
    
    @Override
    public <T> T deserialize(byte[] data, Class<T> clazz) throws Exception {
        return null;
    }
    
    @Override
    public String getContentType() {
        return "application/msgpack";
    }
}

public class SerializationFactory {
    
    private final Map<String, MessageSerializer> serializers = new HashMap<>();
    
    public SerializationFactory() {
        serializers.put("application/json", new JsonMessageSerializer());
        serializers.put("application/msgpack", new MessagePackSerializer());
    }
    
    public MessageSerializer getSerializer(String contentType) {
        return serializers.getOrDefault(contentType, serializers.get("application/json"));
    }
    
    public void registerSerializer(String contentType, MessageSerializer serializer) {
        serializers.put(contentType, serializer);
    }
}

三、代码示例 ​

3.1 完整序列化示例 ​

java
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.rabbitmq.client.*;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.util.*;

public class CompleteSerializationDemo {
    
    private final Channel channel;
    private final ObjectMapper objectMapper;
    private final SerializationFactory serializationFactory;
    
    public CompleteSerializationDemo(Channel channel) {
        this.channel = channel;
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());
        this.serializationFactory = new SerializationFactory();
    }
    
    public void publishMessage(String exchange, String routingKey, 
                              Object message, String messageType) throws Exception {
        
        byte[] body = objectMapper.writeValueAsBytes(message);
        
        Map<String, Object> headers = new HashMap<>();
        headers.put("x-message-type", messageType);
        headers.put("x-serialization", "json");
        headers.put("x-timestamp", System.currentTimeMillis());
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .contentEncoding("UTF-8")
            .messageId(UUID.randomUUID().toString())
            .type(messageType)
            .timestamp(new Date())
            .headers(headers)
            .build();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
    
    public <T> T consumeMessage(Delivery delivery, Class<T> clazz) throws Exception {
        String contentType = delivery.getProperties().getContentType();
        String contentEncoding = delivery.getProperties().getContentEncoding();
        
        if (!"application/json".equals(contentType)) {
            throw new IllegalArgumentException("不支持的内容类型: " + contentType);
        }
        
        String charset = contentEncoding != null ? contentEncoding : "UTF-8";
        String json = new String(delivery.getBody(), charset);
        
        return objectMapper.readValue(json, clazz);
    }
    
    public void demonstrateSerialization() throws Exception {
        String exchange = "test.exchange";
        String routingKey = "test.key";
        
        channel.exchangeDeclare(exchange, BuiltinExchangeType.DIRECT, true);
        channel.queueDeclare("test.queue", true, false, false, null);
        channel.queueBind("test.queue", exchange, routingKey);
        
        OrderMessage order = new OrderMessage("ORD-001", "USER-001", 99.99);
        
        publishMessage(exchange, routingKey, order, "order.created");
        
        GetResponse response = channel.basicGet("test.queue", false);
        if (response != null) {
            OrderMessage received = consumeMessage(response, OrderMessage.class);
            System.out.println("收到订单: " + received);
            
            channel.basicAck(response.getEnvelope().getDeliveryTag(), false);
        }
    }
}

3.2 多格式支持 ​

java
import com.rabbitmq.client.*;
import java.util.*;

public class MultiFormatSerialization {
    
    private final Channel channel;
    private final Map<String, MessageSerializer> serializers;
    
    public MultiFormatSerialization(Channel channel) {
        this.channel = channel;
        this.serializers = new HashMap<>();
        
        serializers.put("json", new JsonMessageSerializer());
        serializers.put("msgpack", new MessagePackSerializer());
    }
    
    public void publish(String exchange, String routingKey, Object message,
                       String format) throws Exception {
        
        MessageSerializer serializer = serializers.get(format);
        if (serializer == null) {
            throw new IllegalArgumentException("不支持的格式: " + format);
        }
        
        byte[] body = serializer.serialize(message);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType(serializer.getContentType())
            .headers(Map.of("x-serialization-format", format))
            .build();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
    
    public <T> T consume(Delivery delivery, Class<T> clazz) throws Exception {
        Map<String, Object> headers = delivery.getProperties().getHeaders();
        String format = headers != null ? 
            (String) headers.get("x-serialization-format") : "json";
        
        MessageSerializer serializer = serializers.get(format);
        if (serializer == null) {
            serializer = serializers.get("json");
        }
        
        return serializer.deserialize(delivery.getBody(), clazz);
    }
}

3.3 带压缩的序列化 ​

java
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.*;
import java.io.*;
import java.util.zip.*;

public class CompressedSerialization {
    
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    public byte[] serializeAndCompress(Object obj) throws Exception {
        byte[] data = objectMapper.writeValueAsBytes(obj);
        
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        try (GZIPOutputStream gzip = new GZIPOutputStream(baos)) {
            gzip.write(data);
        }
        
        return baos.toByteArray();
    }
    
    public <T> T decompressAndDeserialize(byte[] compressedData, 
                                          Class<T> clazz) throws Exception {
        
        ByteArrayInputStream bais = new ByteArrayInputStream(compressedData);
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        
        try (GZIPInputStream gzip = new GZIPInputStream(bais)) {
            byte[] buffer = new byte[1024];
            int len;
            while ((len = gzip.read(buffer)) > 0) {
                baos.write(buffer, 0, len);
            }
        }
        
        return objectMapper.readValue(baos.toByteArray(), clazz);
    }
    
    public void publishCompressed(Channel channel, String exchange, 
                                  String routingKey, Object message) 
            throws Exception {
        
        byte[] body = serializeAndCompress(message);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .contentEncoding("gzip")
            .build();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
    
    public <T> T consumeCompressed(Delivery delivery, Class<T> clazz) 
            throws Exception {
        
        String encoding = delivery.getProperties().getContentEncoding();
        
        if ("gzip".equals(encoding)) {
            return decompressAndDeserialize(delivery.getBody(), clazz);
        } else {
            return objectMapper.readValue(delivery.getBody(), clazz);
        }
    }
}

四、实际应用场景 ​

4.1 微服务消息传递 ​

java
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.*;
import java.util.*;

public class MicroserviceMessage {
    
    private String messageId;
    private String messageType;
    private String sourceService;
    private String targetService;
    private Object payload;
    private Map<String, Object> headers;
    private long timestamp;
    
    public static class Builder {
        private MicroserviceMessage message = new MicroserviceMessage();
        
        public Builder messageId(String messageId) {
            message.messageId = messageId;
            return this;
        }
        
        public Builder messageType(String messageType) {
            message.messageType = messageType;
            return this;
        }
        
        public Builder sourceService(String sourceService) {
            message.sourceService = sourceService;
            return this;
        }
        
        public Builder targetService(String targetService) {
            message.targetService = targetService;
            return this;
        }
        
        public Builder payload(Object payload) {
            message.payload = payload;
            return this;
        }
        
        public Builder headers(Map<String, Object> headers) {
            message.headers = headers;
            return this;
        }
        
        public MicroserviceMessage build() {
            message.messageId = message.messageId != null ? 
                message.messageId : UUID.randomUUID().toString();
            message.timestamp = System.currentTimeMillis();
            return message;
        }
    }
    
    public String getMessageId() { return messageId; }
    public String getMessageType() { return messageType; }
    public String getSourceService() { return sourceService; }
    public String getTargetService() { return targetService; }
    public Object getPayload() { return payload; }
    public Map<String, Object> getHeaders() { return headers; }
    public long getTimestamp() { return timestamp; }
}

class MicroserviceMessagePublisher {
    
    private final Channel channel;
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    public MicroserviceMessagePublisher(Channel channel) {
        this.channel = channel;
    }
    
    public void publish(String exchange, MicroserviceMessage message) 
            throws Exception {
        
        byte[] body = objectMapper.writeValueAsBytes(message);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .messageId(message.getMessageId())
            .type(message.getMessageType())
            .timestamp(new Date(message.getTimestamp()))
            .headers(createHeaders(message))
            .build();
        
        String routingKey = message.getTargetService() + "." + message.getMessageType();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
    
    private Map<String, Object> createHeaders(MicroserviceMessage message) {
        Map<String, Object> headers = new HashMap<>();
        headers.put("x-source-service", message.getSourceService());
        headers.put("x-target-service", message.getTargetService());
        headers.put("x-message-id", message.getMessageId());
        return headers;
    }
}

4.2 事件溯源 ​

java
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.*;
import java.util.*;

public class EventMessage {
    
    private String eventId;
    private String eventType;
    private String aggregateId;
    private String aggregateType;
    private int version;
    private Object data;
    private Map<String, Object> metadata;
    private long timestamp;
    
    public static EventMessage create(String aggregateId, String aggregateType,
                                      String eventType, Object data) {
        EventMessage event = new EventMessage();
        event.eventId = UUID.randomUUID().toString();
        event.eventType = eventType;
        event.aggregateId = aggregateId;
        event.aggregateType = aggregateType;
        event.version = 1;
        event.data = data;
        event.timestamp = System.currentTimeMillis();
        event.metadata = new HashMap<>();
        return event;
    }
    
    public String getEventId() { return eventId; }
    public String getEventType() { return eventType; }
    public String getAggregateId() { return aggregateId; }
    public String getAggregateType() { return aggregateType; }
    public int getVersion() { return version; }
    public Object getData() { return data; }
    public Map<String, Object> getMetadata() { return metadata; }
    public long getTimestamp() { return timestamp; }
}

class EventPublisher {
    
    private final Channel channel;
    private final ObjectMapper objectMapper = new ObjectMapper();
    
    public EventPublisher(Channel channel) {
        this.channel = channel;
    }
    
    public void publishEvent(String exchange, EventMessage event) throws Exception {
        byte[] body = objectMapper.writeValueAsBytes(event);
        
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
            .contentType("application/json")
            .messageId(event.getEventId())
            .type(event.getEventType())
            .timestamp(new Date(event.getTimestamp()))
            .headers(Map.of(
                "x-aggregate-id", event.getAggregateId(),
                "x-aggregate-type", event.getAggregateType(),
                "x-version", event.getVersion()
            ))
            .build();
        
        String routingKey = event.getAggregateType() + "." + event.getEventType();
        
        channel.basicPublish(exchange, routingKey, props, body);
    }
}

五、常见问题与解决方案 ​

5.1 序列化失败 ​

问题描述： 对象序列化时抛出异常。

解决方案：

java
public byte[] safeSerialize(Object obj) {
    try {
        return objectMapper.writeValueAsBytes(obj);
    } catch (Exception e) {
        throw new SerializationException("序列化失败: " + e.getMessage(), e);
    }
}

public <T> T safeDeserialize(byte[] data, Class<T> clazz) {
    try {
        return objectMapper.readValue(data, clazz);
    } catch (Exception e) {
        throw new SerializationException("反序列化失败: " + e.getMessage(), e);
    }
}

5.2 版本兼容性 ​

问题描述： 消息格式变更导致旧消息无法解析。

解决方案：

java
public class VersionedMessage {
    private String version = "1.0";
    private Object data;
    
    public String getVersion() { return version; }
    public Object getData() { return data; }
}

public <T> T deserializeWithVersion(byte[] data, Class<T> clazz) throws Exception {
    VersionedMessage versioned = objectMapper.readValue(data, VersionedMessage.class);
    
    switch (versioned.getVersion()) {
        case "1.0":
            return convertV1ToCurrent(versioned.getData(), clazz);
        case "2.0":
            return objectMapper.convertValue(versioned.getData(), clazz);
        default:
            throw new IllegalArgumentException("不支持的版本: " + versioned.getVersion());
    }
}

5.3 大消息处理 ​

问题描述： 消息过大导致性能问题。

解决方案：

java
public void publishLargeMessage(Channel channel, String exchange, 
                                String routingKey, Object message) throws Exception {
    
    byte[] body = objectMapper.writeValueAsBytes(message);
    
    if (body.length > 1024 * 1024) {
        body = compress(body);
    }
    
    AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
        .contentType("application/json")
        .contentEncoding(body.length > 1024 * 1024 ? "gzip" : "identity")
        .build();
    
    channel.basicPublish(exchange, routingKey, props, body);
}

六、最佳实践建议 ​

6.1 序列化选择 ​

text
选择建议：
├── 通用场景：JSON（可读性好、跨语言）
├── 高性能场景：MessagePack 或 Protobuf
├── 大消息场景：JSON + 压缩
└── Java 专用场景：避免使用 Java 原生序列化

6.2 格式规范 ​

text
格式建议：
├── 统一使用 UTF-8 编码
├── 设置正确的 Content-Type
├── 包含消息类型信息
├── 添加版本号支持兼容性
└── 使用有意义的消息 ID

6.3 安全考虑 ​

text
安全建议：
├── 避免反序列化不可信数据
├── 限制消息大小
├── 验证消息格式
├── 处理异常情况
└── 记录序列化错误

七、相关链接 ​