0

我正在使用 senecajs 开发一个 node.js 社交网络应用程序，并且需要实现一个生产者可以向多个消费者发送相同消息的场景。我发现一篇文章似乎说明了使用 senecajs 完成此操作的示例代码。问题是我正在尝试将其转换为我的场景，这是本文中的示例（https://github.com/senecajs/seneca-amqp-transport/issues/27）：

我有 1 个客户端向 2 个听众发布事件。

客户：

.client({
type: 'amqp',
pin: 'incomingMessage:*',
url: process.env.AMQP_URL,
exchange: {
  name: process.env.NODE_ENV + ':events',
  type: 'fanout'
}
});

听众：

.listen({
type: 'amqp',
pin:  'incomingMessage:*', //maybe useless
url:  process.env.AMQP_URL,
name: process.env.NODE_ENV + ':service1',
exchange: {
  name: process.env.NODE_ENV + ':events',
  type: 'fanout'
}
});

.listen({
type: 'amqp',
pin:  'incomingMessage:*', //maybe useless?
url:  process.env.AMQP_URL,
name: process.env.NODE_ENV + ':service2',
exchange: {
  name: process.env.NODE_ENV + ':events',
  type: 'fanout'
}
});

有几个项目令人困惑：

  1. 对于客户端设置，似乎名称最终会成为“development:events”或“production:events”。我的这种想法正确吗？

  2. 对于exchange对象外的监听器的name字段，这个字段的作用是什么？

  3. 当我调用 add 方法时，我需要传入一个名称，该名称映射到侦听器收到消息时进行的函数调用，我是否会将“incomingMessage：*”传递给 add 调用？

  4. 这段代码真的会使用 senecajs 有效地提供扇出功能吗？

4

1 回答 1

2

我已经测试了样品。（归功于 GitHub 上的作者）

1，通常NODE_ENV环境变量是指开发，生产，登台等。它可以是任何东西（如“apple”），但诀窍是这个“name”属性是指“队列”名称，它必须是唯一的每个听众/订阅者。

2，就像我上面提到的，这个“名称”属性（交换对象之外）是队列的名称。对于每个听众来说，它必须是独一无二的。每个侦听器都有自己的绑定到交换的队列。

3，它可以是任何东西。但这也很棘手。

对于将“发布”到扇出交换的客户端，它需要在选项中具有与被调用操作的模式相同的引脚。例如，如果客户将调用：

seneca.act('event:orderReceived', optionalPayload)

那么引脚必须是：

{
    pin: 'event:orderReceived',
    ...
}

有关模式的工作原理，请参阅Seneca API 文档。

对于 listeners，在这种情况下（对于 RPC 这不是真的）他们不必匹配 pin。当您调用 seneca.add 函数时，只有模式必须匹配（有点奇怪）。例如：

seneca.add('event:orderReceived', function (msg, respond) {
    // some logic
})

pin 属性可以是：

{
     pin: 'something:different'
}

4、对于最后一个问题的答案是正确的，它可以提供发布/订阅功能。我已经测试过了。

请注意，如果您在“发布”交换时没有将回调传递给 seneca.act，那么它不会等待任何响应。例如：

seneca.act('event:orderReceived', { some: "payload" })

另请注意，如果您不需要响应，则null无论如何都必须在侦听器服务中使用 for first (error) 参数调用响应回调，否则它将超时。请参阅下面的示例。

这是我的代码版本：

出版商

const seneca = require('seneca')
const si = seneca()

si
    .use('seneca-amqp-transport')

    .client({
        type: 'amqp',
        pin: 'event:orderReceived',
        url: 'amqp://localhost:5672',
        exchange: {
            name: 'order-service',
            type: 'fanout'
        }
    })
    .ready( function() {
        si.log.info('order-service application is running...')
        si.act('event:orderReceived', {})
    })

process.on('SIGTERM', () => {
    si.log.info('Got SIGTERM. Graceful shutdown start')
    si.act('role:seneca,cmd:close')
})

订户 1

const seneca = require('seneca')
const si = seneca()

si
    .use('seneca-amqp-transport')

    .add('event:orderReceived', function(msg, respond) {
        si.log.info(msg)
        respond(null)
    })

    .listen({
        type: 'amqp',
        pin:  'some:pin1',
        url:  'amqp://localhost:5672',
        name: 'delivery-service-queue',
        exchange: {
            name: 'order-service',
            type: 'fanout'
        }
    })

    .ready( function() {
        si.log.info('delivery-service application is running...')
    })

process.on('SIGTERM', () => {
    si.log.info('Got SIGTERM. Graceful shutdown start')
    si.act('role:seneca,cmd:close')
})

订户 2

const seneca = require('seneca')
const si = seneca()

si
    .use('seneca-amqp-transport')

    .add('event:orderReceived', function(msg, respond) {
        si.log.info(msg)
        respond(null)
    })

    .listen({
        type: 'amqp',
        pin:  'some:pin2',
        url:  'amqp://localhost:5672',
        name: 'financial-service-queue',
        exchange: {
            name: 'order-service',
            type: 'fanout'
        }
    })

    .ready( function() {
        si.log.info('financial-service application is running...')
    })

process.on('SIGTERM', () => {
    si.log.info('Got SIGTERM. Graceful shutdown start')
    si.act('role:seneca,cmd:close')
})

我不确定这是一种正确的方法，但似乎可行。

于 2017-06-09T20:04:47.200 回答