标签: message-queue

我想使用 windows 的消息队列工具将结构发送到另一个线程。但是我发现postthreadmessage函数只提供了两个整型参数，lparam和wparam给我传递参数，所以我决定把struct的地址放在lparam里面。这是 Windows 用来传递结构的正确方式吗？

我打算使用 boost::shared_ptr 在接收者线程和发送者线程中保存 struct 的地址。我怀疑当两个 shared_ptrs 超出范围时，结构会被释放两次吗？我想不出一种方法来确保在堆上分配的结构将被 100% 释放，有什么想法吗？

注意：在Symfony2 中通过RabbitMQBundle使用RabbitMq。

我的制作人发送这样的消息：

$message = array(
    'class' => get_class($receiver),
    'id' => $receiver->getId(),
    'stepNumber' => 1,
    'errorCount' => 0
);

消费者$receiver从数据库中检索并给他发送一封电子邮件。

public function execute(AMQPMessage $msg)
{
    //Step1 - retrieve user from db

    //Step2 - send email

    //Step3 - update stuff in database
}

为了跟踪错误，我想在每一步处理异常。如果在第3步抛出异常，我想修改stepNumber为3，将errorCountin增加1 $msg，最后$msg通过返回重新排队false。

这具有以下优点：

• 当消费者再次处理消息时，它不会再次发送电子邮件。
• 当errorCount > 5，我只是丢弃消息.. return false。

这会很棒，但是：

有没有办法$msg在RabbitMQ重新排队之前修改它？

我正在使用一个使用受限的 API，比方说：每秒不超过 10 次调用，每天不超过 5000 次调用。

我正在 beanstalkd 队列进程作业中处理此调用。考虑到 API 的限制，我如何限制此作业的处理。

我正在尝试从特权进程（等待稍后读取）创建 POSIX 消息队列，然后从非特权进程（发送消息）打开此消息队列，然后返回 mq_open()：EACCES。

如果创建进程和打开进程都有特权或都没有特权，则 mq_open 将成功。

我检查了 mq_open 手册，它说 EACCES 意味着调用者没有在指定模式下打开它的权限，但我不确定什么是“指定模式”......

在特权进程中创建成功：

struct mq_attr attr;
attr.mq_flags = O_RDONLY;
attr.mq_maxmsg = 10;
attr.mq_msgsize = 1024;
attr.mq_curmsgs = 0;

mq_open("/myMq", (O_RDONLY| O_CREAT), (S_IRWXU | S_IRWXG | S_IRWXO) /* 777 */, &attr);

在非特权进程中打开失败：

mqd_t mqd;
mqd = mq_open("/myMq", (O_WRONLY|O_NONBLOCK));
if ((mqd_t)-1 == mqd) {
    printf("mq_open %d %s\n", errno, strerror(errno)); }

它给出了错误：mq_open 13 Permission denied

我使用的是 CentOS 6.5 64 位

uname -r
2.6.32-431.el6.x86_64

你能帮我找出问题所在。提前致谢。

我们使用 Rebus 作为 Sql 服务器的队列系统。对于不同类型的消息，我们有多个收件人。每条消息都可以由特定类型的多个工作人员处理。一条消息只能由一名工人处理/处理（第一个拉出它的工人）。如果工作人员由于某种原因无法完成它，它会使用超时服务推迟消息。

如果我理解正确，它就会变成一个 TimeoutRequest 并放入超时表中。当需要重新运行时，它会在作为原始消息重新引入队列之前变成 TimeoutReply。

我们遇到的问题是，当它变成 TimeoutReply 时，所有工作人员都会拿起它并创建原始消息。当超时时，一条原始消息变成多条消息（与工作人员一样多）。

我们的 Rebus 设置如下：

“服务器端”：

        var adapter = new BuiltinContainerAdapter();
        Configure.With(adapter)
            .Logging(l => l.Log4Net())
            .Transport(t => t.UseSqlServerInOneWayClientMode(connectionString).EnsureTableIsCreated())
            .CreateBus()
            .Start();

        return adapter;

“工人方面”：

        _adapter = new BuiltinContainerAdapter();
        Configure.With(_adapter)
            .Logging(l => l.Log4Net())
            .Transport(t => t.UseSqlServer(_connectionString, _inputQueue, "error")
                .EnsureTableIsCreated())
            .Events(x => x.AfterMessage += ((bus, exception, message) => SendWorkerFinishedJob(exception, message)))
            .Events(x => x.BeforeMessage += (bus, message) => SignalWorkerStartedJob(message))
            .Behavior(x => x.SetMaxRetriesFor<Exception>(0))
            .Timeouts(x => x.StoreInSqlServer(_connectionString, "timeouts").EnsureTableIsCreated())
            .CreateBus().Start(numberOfWorkers);

非常感谢在解决问题或提供理解方面的任何帮助！

我对 RabbitMQ 很陌生，我需要编写一个程序，该程序具有单个队列的多生产者和多消费者。这可以做我在图片中显示的内容吗？我找到了很多例子，但它们都是单一生产者。任何生产者向任何消费者发送消息。

我是 RabbitMQ 的新手，我刚刚浏览了 Rabbitmq 文档（路由）。我在 Exchange 与路由键之间很困惑。我的要求是，我想动态创建多个队列。请参考下图。

前任。让我们说如果生产者为消费者 c3 创建消息，那么它应该转到 Exchange 并仅路由到队列 3 并仅由 C3 消费。目前我只需要 3 个队列，将来这个计数可能会增加。那么如何处理这种情况呢。

注意：我参考这个博客交流

我已经将 Spring hibernate 与 Rabbitmq 一起使用了。下面的代码显示了 Rabbit MQ 侦听器配置文件。

<?xml version="1.0" encoding="UTF-8" ?>
<beans xmlns="http://www.springframework.org/schema/beans" 
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" 
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans 
http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
http://www.springframework.org/schema/rabbit  
http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd">

<rabbit:connection-factory id="connectionFactory" host="xx.xx.10.181" username="admin" password="admin" />
<rabbit:admin connection-factory="connectionFactory" />

<!--  Create Queues -->
<rabbit:queue id="q1" name="q1.queue" durable="true" />
<rabbit:queue id="q2" name="q2.queue" durable="true"  />
<rabbit:queue id="q3" name="q3.queue" durable="true"  />

<!--create Exchange here -->
<rabbit:direct-exchange id="myExchange" name="MY Exchange">
    <rabbit:bindings>
         <rabbit:binding queue="q1" …

我有一个风暴拓扑来处理来自 Kafka 的消息，并根据手头的任务在 Cassandra 中进行 HTTP 调用/保存。我一收到消息就处理。由于响应来自外部源（如 HTTP），很少有消息没有被完全处理。我想为重试实现指数退避机制，以防 HTTP 服务器在一段时间后不响应/返回错误消息以重试。我可以想到一些可以实现它们的想法。如果我可以使用任何其他容错解决方案，我想知道其中哪一个是更好的解决方案。由于这用于实现指数退避，因此每条消息将具有不同的延迟时间。

• 将Kafka中的另一个主题发送给它，稍后再使用。我首选的解决方案。我知道我们可以使用 Kafka 偏移量，以便在后期使用消息。我怎么找不到文档/示例代码来做同样的事情。如果有人可以帮助我解决这个问题，那将非常有帮助。
• 编写消息Cassandra / Redis并编写调度程序以获取未处理且准备好使用的消息并将其发送到 Kafka，以便我的 Storm 拓扑可以使用它。（其他遗留项目中的现有解决方案（非风暴））
• 延迟发送到Beanstalk（其他遗留项目中的现有解决方案（非风暴）。我想避免使用这个解决方案，只有在我无法选择的情况下才使用它）。

虽然这几乎是我想做的。我无法找到文档来实现Kafka - Delayed Queue implementation using high level consumer 中提到的delayProcessingUntil

我过去曾使用 Beanstalk 从 Data-store 完成预定工作并延迟，但我更喜欢使用 Kafka。

我正在尝试建立一个API，该API使用另一台服务器上的队列系统来处理请求。让我开始尝试在没有队列系统的情况下要完成的工作（无权使其保持简单）：例如，使用Postman对URL https://example.com/products进行GET请求，将返回JSON字符串，例如

[
    {
        "id": 1,
        "name": "some name",
        ...
    },
    {
        "id": 2,
        "name": "some other name",
        ...
    }.
    ...
]

route / api.php中的代码将类似于：

<php

Route::get('/products', ProductController@index');

以及app / Http / Controllers / ProductController.php中的代码：

<?php

namespace App\Http\Controllers;

class ProductController extends Controller
{
    /**
     * Return the products.
     *
     * @return \Illuminate\Http\Response
     */
    public function index()
    {
        // Logic to get the products.
        return $products->toJson();
    }
}

我要完成的工作是，所有业务逻辑都在另一台运行多个工作程序的服务器上进行处理。以下是我对此的推理。

• 安全性：万一我们被黑，它很可能是客户端服务器，而不是辅助服务器。由于最后一个具有所有业务逻辑，因此在最坏的情况下，黑客将只能获取传入和传出的请求数据。
• 多个工作人员：获得产品可能不会花费很长时间，但是可能还有其他要求需要更多时间来处理。在大多数情况下，发出请求的用户将必须等待结果。但是，其他拨打电话的用户不必等待。因此，另一名工人可以接受此请求并处理工作。

这就是我所看到的工作流程：

• 所有的免费工人都在不停地轮询队列中的工作

  1. 用户提出要求
  2. 客户端服务器接收请求数据并将其放在队列中
  3. 工人从队列中拿出一份工作并处理
  4. 工作程序将结果返回到客户端服务器
  5. 客户端服务器将结果返回给用户

下面是一幅小图，用于清除一切。

  User …

我已经阅读了一些讨论和论坛，但仍然无法理解哪种产品更胜于rabbitmq vs activemq。我听说Rabbitmq的延迟更多，因为所有流量首先都转移到了中央节点。如果有人可以向我指出一些文档，可以突出显示在哪种情况下应使用的文档，那将是很好的。谢谢大家