如何基于发布者确认提供响应

Question

我有一个web服务，它接收对象，通过AMQP发送通知，并向请求者返回JSON响应。每个请求都是在一个线程上执行的，我正在尝试实现publisher confirms，我正在努力解决如何设置它。我有它的工作，但我不喜欢我这样做。

我这样做的方式是：

• 在邮件上添加一些标题
• 拥有一个包含2个订阅者的发布-订阅频道
• 订户1）创建一个阻塞队列，使其准备就绪，并通过amqp发送消息
• 订户2）开始在该队列上拉动5秒钟，直到得到确认
• amqp：出站通道适配器将其发布者确认发送给服务激活器
• PublisherConfigReceiver接收确认并将其放入阻塞队列，导致订阅服务器2的拉取完成并返回确认结果

这种技术确实可以正常工作，但我不喜欢假设该链将在来自发布-订阅通道的WaitForPublisherConfig服务激活器之前接收消息。在这种情况下，顺序关系到哪个组件首先接收消息。

如果waitForPublisher确认服务激活器首先接收到消息，它只会阻塞线程5秒钟，然后允许链通过amqp： outbind-Channel-适配器发送消息。

我尝试将waitForPublisherConders放在amqp： outbinding-Channel适配器之后，但是由于outbinding-Channel适配器不会“返回”任何东西，所以服务激活器永远不会在链中被调用。

我觉得应该有更好的方法。我的目标是等待publisher确认（或者在spring的publisher确认中找不到支持的超时），然后再向请求者发送响应。

您是否可以帮助我更好地制定解决方案，或者让我知道是否可以依赖发布-订阅频道的第一个订户始终首先接收消息这一事实。

对不起，这个太长了。

一些配置

<int:header-enricher input-channel="addHeaders" output-channel="metadataIngestNotifications">
    <int:header name="routingKey" ref="routingKeyResolver" method="resolveReoutingKey"/>
    <int:header name="notificationId" expression="payload.id" />
</int:header-enricher>

<int:chain input-channel="metadataIngestNotifications" output-channel="nullChannel" >

    <int:service-activator id="addPublisherConfirmQueue"
        requires-reply="false"
        ref="publisherConfirmService"  
        method="addPublisherConfirmQueue" />

    <int:object-to-json-transformer id="transformObjectToJson" />

    <int-amqp:outbound-channel-adapter id="amqpOutboundChannelAdapter"
        amqp-template="rabbitTemplate"
        exchange-name="${productNotificationExchange}"
        confirm-ack-channel="publisherConfirms"
        confirm-nack-channel="publisherConfirms"
        mapped-request-headers="*"
        routing-key-expression="headers.routingKey"
        confirm-correlation-expression="headers.notificationId" />

</int:chain>

<int:service-activator id="waitForPublisherConfirm"
        input-channel="metadataIngestNotifications"
        output-channel="publisherConfirmed"
        requires-reply="true"
        ref="publisherConfirmService"  
        method="waitForPublisherConfirm"  />


<int:service-activator id="publisherConfirmReceiver" 
                       ref="publisherConfirmService" 
                       method="receivePublisherConfirm" 
                       input-channel="publisherConfirms" 
                       output-channel="nullChannel" />

班

public class PublisherConfirmService {

    private final Map<String, BlockingQueue<Boolean>> suspenders = new HashMap<>();

    public Message addPublisherConfirmQueue(@Header("notificationId") String id, Message m){
        LogManager.getLogger(this.getClass()).info("Adding publisher confirm queue.");
        BlockingQueue<Boolean> bq = new LinkedBlockingQueue<>();
        suspenders.put(id, bq);
        return m;
    }

    public boolean waitForPublisherConfirm(@Header("notificationId") String id) {
        LogManager.getLogger(this.getClass()).info("Waiting for publisher confirms for Notification: " + id);
        BlockingQueue<Boolean> bq = suspenders.get(id);
        try {
            Boolean result = bq.poll(5, TimeUnit.SECONDS);
            if(result == null){
                LogManager.getLogger(this.getClass()).error("The broker took too long to return a publisher confirm. NotificationId: " + id);
                return false;
            }else if(!result){
                LogManager.getLogger(this.getClass()).error("The publisher confirm indicated that the message was not confirmed. NotificationId: " + id);
                return false;
            }
        } catch (InterruptedException ex) {
            LogManager.getLogger(this.getClass()).error("Something went wrong polling for the publisher confirm for notificationId: " + id, ex);
            return false;
        }finally{
            suspenders.remove(id);
        }
        return true;
    }

    public void receivePublisherConfirm(String id, @Header(AmqpHeaders.PUBLISH_CONFIRM) boolean confirmed){
        LogManager.getLogger(this.getClass()).info("Received publisher confirm for Notification: " + id);
        if (suspenders.containsKey(id)){
            BlockingQueue<Boolean> bq = suspenders.get(id);
            bq.add(confirmed);
        }
    } 
}

Answer 1

出于同样的目的，我们来看看聚合器解决方案怎么样？

<代码>

confirm ack channel必须是经过某种转换后将消息带到同一聚合器的内容，例如正确提取correlationKey等。