// 生产者 - 添加消息
MessagePipe pipe = messagePipeManager.getMessagePipe("order-processing");
Message message = new Message("order-123".getBytes());
pipe.putLast(message);  // 线程安全地添加消息

// 客户端 - 处理消息
@Component
public class OrderProcessor implements MessageProcessor {
    @Override
    public String bindingPipeName() {
        return "order-processing";  // 绑定到指定管道
    }

    @Override
    public boolean processing(String pipeName, String requestId, Message message) {
        // 顺序处理订单
        String orderId = new String(message.getBody());
        processOrder(orderId);
        return true;  // 返回 true 表示成功
    }
}

@Bean
public MessagePipeConfiguration messagePipeConfiguration() {
    return MessagePipeConfiguration.defaultConfiguration()
        // 重试记录在 Redis 中保留 30 天
        .setDlqMessageExpireSeconds(30 * 24 * 60 * 60)
        // 其他配置...
        ;
}

class DeadLetterRecord {
    String messageId;           // 消息 ID
    byte[] messageBody;         // 原始消息内容
    String failureReason;       // 失败原因
    int retryCount;            // 重试次数
    long expireTime;           // 过期时间
    LocalDateTime createTime;  // 创建时间
}

// 查询死信队列中的消息
List<DeadLetterRecord> records =
    deadLetterQueue.listMessages("order-processing");

// 手动恢复消息到主队列
deadLetterQueue.recoverMessage("order-processing", messageId);

// 清空死信队列
deadLetterQueue.clear("order-processing");

@Bean
public ClientConfiguration clientConfiguration() {
    return new ClientConfiguration()
        .setLocalPort(5201)
        .setServerAddress("localhost")
        .setServerPort(5200)
        .setHeartBeatIntervalSeconds(10)    // 心跳间隔
        .setRetryRegisterTimes(3);          // 注册重试次数
}

@Component
public class CustomLoadBalanceStrategy implements ClientLoadBalanceStrategy {
    @Override
    public ClientInformation select(String pipeName, List<ClientInformation> clients) {
        // 实现自己的负载均衡逻辑
        // 例如：轮询、最少连接、一致性哈希等
        return clients.get(new Random().nextInt(clients.size()));
    }
}

@Component
public class OrderProcessor implements MessageProcessor {
    @Override
    public String bindingPipeName() {
        return "order-processing";  // 精确匹配这个管道名
    }

    @Override
    public ProcessorType processorType() {
        return ProcessorType.SPECIFIC;
    }

    @Override
    public boolean processing(String pipeName, String requestId, Message message) {
        // 处理 order-processing 管道的消息
        return true;
    }
}

@Component
public class RegexProcessor implements MessageProcessor {
    @Override
    public String bindingPipeName() {
        return "order-.*";  // 正则表达式
    }

    @Override
    public ProcessorType processorType() {
        return ProcessorType.REGEX;
    }

    @Override
    public boolean processing(String pipeName, String requestId, Message message) {
        // 处理所有匹配 order-.* 的管道
        System.out.println("Processing pipe: " + pipeName);
        return true;
    }
}

class PipeMetrics {
    long totalInputCount;     // 总输入消息数
    long totalProcessCount;   // 总处理消息数
    int currentQueueSize;     // 当前队列大小
    List<ClientInformation> boundClients;  // 绑定的客户端
}

// 获取单个管道指标
MessagePipeMetrics metrics = pipe.getMetrics();
System.out.println("输入: " + metrics.getTotalInputCount());
System.out.println("处理: " + metrics.getTotalProcessCount());

// 获取所有管道的聚合指标
MessagePipeMetricsAggregator aggregator = new MessagePipeMetricsAggregator();
long totalInput = aggregator.aggregateInputCount();
long totalProcess = aggregator.aggregateProcessCount();

@SpringBootApplication
@EnableMessagePipeServer  // 启用 Server 端功能
public class ServerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ServerApplication.class, args);
    }
}

@SpringBootApplication
@EnableMessagePipeClient  // 启用 Client 端功能
public class ClientApplication {
    public static void main(String[] args) {
        SpringApplication.run(ClientApplication.class, args);
    }
}

// 生产者代码
MessagePipe pipe = messagePipeManager.getMessagePipe("order-process");
Message msg = new Message("order-123".getBytes());
pipe.putLast(msg);  // 或 putLastBatch(List) 批量添加

@SpringBootApplication
@EnableMessagePipeServer  // 启用服务端
public class MessagePipeServerApplication {
    public static void main(String[] args) {
        SpringApplication.run(MessagePipeServerApplication.class, args);
    }
}

@RestController
@RequestMapping("/api/messages")
public class MessageController {

    @Autowired
    private DefaultMessagePipeManager messagePipeManager;

    @PostMapping("/send")
    public void sendMessage(@RequestParam String pipeName,
                           @RequestParam String messageBody) {
        MessagePipe pipe = messagePipeManager.getMessagePipe(pipeName);
        Message message = new Message(messageBody.getBytes());
        pipe.putLast(message);
    }

    @PostMapping("/send-batch")
    public void sendBatch(@RequestParam String pipeName,
                         @RequestBody List<String> messages) {
        MessagePipe pipe = messagePipeManager.getMessagePipe(pipeName);
        List<Message> msgList = messages.stream()
            .map(body -> new Message(body.getBytes()))
            .collect(Collectors.toList());
        pipe.putLastBatch(msgList);
    }
}

@SpringBootApplication
@EnableMessagePipeClient  // 启用客户端
public class MessagePipeClientApplication {
    public static void main(String[] args) {
        SpringApplication.run(MessagePipeClientApplication.class, args);
    }
}

@Component
public class OrderMessageProcessor implements MessageProcessor {

    private static final Logger logger = LoggerFactory.getLogger(OrderMessageProcessor.class);

    @Override
    public String bindingPipeName() {
        return "order-processing";
    }

    @Override
    public ProcessorType processorType() {
        return ProcessorType.SPECIFIC;
    }

    @Override
    public boolean processing(String pipeName, String requestId, Message message) {
        try {
            String orderId = new String(message.getBody());
            logger.info("Processing order: {}, requestId: {}", orderId, requestId);

            // 您的业务逻辑
            processOrder(orderId);

            return true;  // 返回 true 表示处理成功
        } catch (Exception e) {
            logger.error("Failed to process order", e);
            return false;  // 返回 false 表示处理失败，会触发重试
        }
    }

    private void processOrder(String orderId) {
        // 处理订单逻辑
        // 例如：数据库更新、调用下游服务等
    }
}

@Component
public class PaymentMessageProcessor implements MessageProcessor {

    @Override
    public String bindingPipeName() {
        return "payment-.*";  // 匹配所有 payment-* 的管道
    }

    @Override
    public ProcessorType processorType() {
        return ProcessorType.REGEX;
    }

    @Override
    public boolean processing(String pipeName, String requestId, Message message) {
        String paymentId = new String(message.getBody());
        System.out.println("Processing payment: " + paymentId +
                          " from pipe: " + pipeName);
        // 处理支付逻辑
        return true;
    }
}

@GetMapping("/metrics")
public ResponseEntity<?> getMetrics() {
    MessagePipe pipe = messagePipeManager.getMessagePipe("order-processing");
    return ResponseEntity.ok(Map.of(
        "totalInput", pipe.getTotalInputCount(),
        "totalProcess", pipe.getTotalProcessCount(),
        "currentQueueSize", pipe.getCurrentQueueSize(),
        "boundClients", pipe.getBoundClients()
    ));
}

@PostMapping("/dlq/recover")
public void recoverFromDLQ(@RequestParam String pipeName,
                          @RequestParam String messageId) {
    MessageDeadLetterQueue dlq = deadLetterQueueManager.getQueue(pipeName);
    dlq.recoverMessage(messageId);
}

@GetMapping("/dlq/list")
public List<DeadLetterRecord> listDLQ(@RequestParam String pipeName) {
    MessageDeadLetterQueue dlq = deadLetterQueueManager.getQueue(pipeName);
    return dlq.listMessages();
}

@Component
public class ConsistentHashStrategy implements ClientLoadBalanceStrategy {
    @Override
    public ClientInformation select(String pipeName,
                                   List<ClientInformation> clients) {
        // 一致性哈希实现
        // 确保同一消息始终发往同一客户端
        return clients.get(hashFunction(pipeName) % clients.size());
    }
}

// 推荐做法
byte[] messageBody = referenceId.getBytes();  // "oss://file-id-123"
Message message = new Message(messageBody);
pipe.putLast(message);

// 客户端处理
public boolean processing(..., Message message) {
    String fileId = new String(message.getBody());
    byte[] actualContent = fetchFromOss(fileId);
    // 处理实际内容
    return true;
}

@Component
public class OrderProcessor implements MessageProcessor {

    @Override
    public boolean processing(String pipeName, String requestId, Message message) {
        String orderId = new String(message.getBody());

        try {
            // 业务逻辑（应该是幂等的）
            boolean success = processOrderIdempotent(orderId);
            return success;
        } catch (Exception e) {
            logger.error("Error processing order: {}", orderId, e);
            return false;  // 返回 false 触发重试
        }
    }

    // 实现幂等性处理
    private boolean processOrderIdempotent(String orderId) {
        // 检查是否已处理过
        Order order = orderRepository.findById(orderId);
        if (order != null && order.isProcessed()) {
            return true;  // 已处理过，返回成功
        }

        // 处理订单
        order.setProcessed(true);
        orderRepository.save(order);
        return true;
    }
}

// 阻塞过长
public boolean processing(..., Message message) {
    Thread.sleep(60000);  // 不要阻塞！
    return true;
}

// 抛出异常
public boolean processing(..., Message message) {
    throw new RuntimeException("Error");  // 应该返回 false
}

// 非幂等
public boolean processing(..., Message message) {
    // 无法判断是否已处理，重试会重复处理
    deductFromAccount(accountId, amount);
    return true;
}

public boolean processing(..., Message message) {
    try {
        // 业务处理
        processMessage(message);
        return true;
    } catch (DatabaseException e) {
        logger.warn("Database error, will retry", e);
        return false;  // 数据库错误 → 重试
    } catch (ValidationException e) {
        logger.error("Invalid message, move to DLQ", e);
        return true;  // 验证失败 → 不重试（虽然会被删除）
    }
}

@Bean
public MessagePipeConfiguration messagePipeConfiguration() {
    return MessagePipeConfiguration.defaultConfiguration()
        .setBatchSize(500);  // 增加批大小提高吞吐，但增加延迟
}

// 更激进的重试（更快恢复）
.setRetryRecordExpireSeconds(24 * 60 * 60)  // 1 天

// 更保守的重试（减少告警）
.setRetryRecordExpireSeconds(7 * 24 * 60 * 60)  // 7 天

// 部署多个客户端实例，绑定同一管道
// 服务端会自动负载均衡分配消息