我将开发一个涉及数据采集服务器的系统，其中每次采集都会填充一行。我还需要能够通知用户应用程序何时获取新数据。
根据我的阅读，使用数据库作为消息队列不是一个好主意，反之亦然，但我想知道是否可以同时使用两者？
采集应用程序可以将新行添加到数据库中，然后通知消息系统中的侦听器。这是否是此类系统的最佳方法？会不会太复杂了？是否有一种设计模式已经实现了这一点？
我不是程序员，但我试图通过给他们一些指导来帮助他们。我们不再拥有任何有关 msmq 的内部专业知识。我们正在尝试使用它来将一些功能与调度应用程序集成。
调度应用程序通过使用自定义构建的 dll 进行网络调用来启动作业。dll 调用 weburl。Web 应用程序将运行其任务并向网站发送有关其执行的任务的更新。网站将消息写入队列。调用该站点的 dll 正在监视队列中是否有带有分配给该作业的标签的消息。当它收到最终状态消息时，它会关闭。
我们每隔几个小时就会收到以下消息。我们每小时运行近 100 个使用此方法的作业。在底部列出的代码中，jobid 对应于消息队列中消息的标签。每个作业在开始时都会发出一个 jobid，并将使用它作为发送到该作业的 msmq 的每条消息的标签。
System.Messaging.MessageQueueException (0x80004005): Message that the cursor is currently pointing to has been removed from the queue by another process or by another call to Receive without the use of this cursor.
at System.Messaging.MessageQueue.ReceiveCurrent(TimeSpan timeout, Int32 action, CursorHandle cursor, MessagePropertyFilter filter, MessageQueueTransaction internalTransaction, MessageQueueTransactionType transactionType)
at System.Messaging.MessageEnumerator.get_Current()
这是它的代码。
while ( running )
{
// System.Console.WriteLine( "Begin Peek" );
messageQueue.Peek();
//System.Console.WriteLine( "End Peek" ); …有没有免费的解决方案可以在 Delphi 中使用消息队列？
我只知道Habari（不是免费的），但在互联网上找不到免费的解决方案。
关于结束我的主题的投票。我在问一个问题，如果我需要在这里解释它是什么，可能我不知道如何回答。知道我在说什么的人可以毫无问题地回答，比如 Daniele 和 Jeroen Pluimers。
但对于那些不知道什么是 MQ 的人：在这里您可以了解它的基础知识：http://mq.java.net/overview.html 了解它是什么之后，您可以阅读有关代理的信息：ActiveMQ， HornetQ、GlassFish、RabbitMQ 等...现在您知道，我们需要一个客户端与经纪人交谈，这就是我正在寻找的客户端，在这种情况下，选项之一（几个选项）是http:// /www.habarisoft.com/habari_openmq.html。
塔克斯
我正在使用 Django、Celery 和 RabbitMQ 编写一个邮件列表管理器。当收到消息时，将为每个收件人执行一项任务。所有任务都进入一个队列，一个或多个工作人员使用队列中的任务，构建电子邮件消息并发送它们。
单个队列会导致公平问题：如果邮件进入大型邮件列表，则会将大量任务添加到队列中，并且其他邮件无法到达其他较小的邮件列表，直到所有邮件都到达大型列表已经发出了。我怎样才能找到解决这个问题的方法？
从概念上讲，解决方案是为每个邮件列表创建一个队列，并让工作人员从各个队列循环中消耗任务。鉴于我需要能够动态创建新的邮件列表，这在 Celery 中可能吗？我还没有看到动态创建队列或使工作人员从新队列消费的功能。
RabbitMQ 有什么方法可以让多个消费者从同一个队列获取相同的消息吗？
我需要向任何正在收听的人发送同样的信息，但也要确保有人处理它。基本上，我需要交换器的扇出功能与队列的basic.ack功能相结合。有什么方法可以以可扩展的方式实现这一点吗？
我正在尝试编写一个客户端和服务器，它们将使用 POSIX 消息队列交换数据。我尝试通过查看在互联网上看到的示例和课程文档来做到这一点。但是，我陷入困境。当我运行它时，我收到“打开文件太多”错误。这是我的代码：
客户：
int main( int argc, char *argv[]) {
//Open its queue, which is client queue
char cq_name[10];
sprintf( cq_name, "/cq%i", getpid());
printf( "Client Queue name: %s\n", cq_name);
mqd_t cq_id = mq_open( cq_name, O_CREAT | O_RDWR, 0666, NULL);
if( cq_id == -1) {
printf( "Error in cq: %s\n", strerror( errno));
return -1;
}
printf( "Name: %s\n", argv[1]);
//Connect to the server message queue
mqd_t sq_id = mq_open( argv[1], O_RDWR);
if( sq_id == -1) {
printf( "Error in sq: …我在 SQS 队列中有一批n消息和一些工作人员。这些工作人员从队列中获取消息，对其进行处理，如果成功则将其删除。当所有工人完成这批工作后n消息后，我想执行一项附加操作。唯一的问题是确定批次何时完成。
一种方法是检查队列是否为空。当我查看 SQS API 时，唯一看起来接近的是ApproximateNumberOfMessages您从GetQueueAttributes. 然而，“大约”这个词表明它实际上并不是我想要的，它的目的更多的是根据队列中大约有多少消息来增加和减少工作人员的数量。
实现我想要的目标的标准方法是什么？或者 SQS 不适合这个目的？
我正在使用一个实现了两个 posix 函数的系统
mq_timedreceive() and pthread_cond_timedwait()
这两个函数都使用基于 CLOCK_REALTIME 的绝对超时。该时钟在系统启动过程中的不同时间会发生变化，并且可以向后或向前移动 10 秒到几小时。
开放组说道：
如果支持Timers选项，则超时应基于CLOCK_REALTIME时钟；如果不支持 Timers 选项，则超时应基于 time() 函数返回的系统时钟。(mq_timedreceive)
对于操作员不连续地提前系统时钟的情况，预计实现会处理在中间时间到期的任何定时等待，就好像该时间实际上已经发生一样。(pthread_cond_timedwait())。
然而，这对于时钟向后设置的情况并没有提供任何指导。
QNX 通过提供解决了这个问题
mq_timedreceive_monotonic().
mq_timedreceive_monotonic() 函数是 QNX Neutrino 扩展；它与 mq_timedreceive() 类似，但它使用 CLOCK_MONOTONIC，因此超时不受系统时间更改的影响。
有没有好的方法在linux中实现QNX功能？
对于 mq_timedreceive()，我可以使用 mq_receive() 和 poll()。但对于条件变量我还没有想出一个干净的方法。我可以使用计时器和信号，但这似乎太复杂了。
当然，另一个解决方案是不调整时钟或使用从 CLOCK_MONOTONIC 派生的不同时钟，但我没有自由更改设计。
我正在尝试获取具有特定相关 ID 的消息，如rabbitmq 文档中所述。但是我看到不相关的消息被出队。我不希望它发生。在我收到消息并知道这不是我要找的消息后，如何告诉rabbitmq不要出队。请帮我。
`
.
.
replyQueueName = channel.queueDeclare().getQueue();
consumer = new QueueingConsumer(channel);
channel.basicConsume(replyQueueName, false, consumer);
while (true) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
System.out.println(delivery.getProperties().getCorrelationId());
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
response = new String(delivery.getBody());
break;
}
}
`
假设我们有以下场景：
如果我尽可能快地将这 10k 条消息排队，则外部系统将无法处理并发负载。我不知道Azure 同时运行多少个函数实例或应用服务实例（“横向扩展”）来处理这些消息。
这是我的函数定义的代码示例：
public class MyQueueProcessor
{
private IMyDependency MyDependency { get; }
public MyQueueProcessor(IMyDependency myDependency)
{
MyDependency = myDependency;
}
[FunctionName(nameof(MyQueueProcessor))]
public async Task Run([ServiceBusTrigger("my-queue-name", Connection = "MyQueueConnection")] string item, ILogger log)
{
var req = JsonConvert.DeserializeObject<MyQueueRequest>(item);
log.LogInformation($"Beginning processing of " + req);
// Valudate req
// Call external service REST API
await MyDependency.CallService(req, new { otherParameters = …