标签: message-queue

给定:我在页面加载时加载了一个消息队列.

使用jGrowl,我如何以类似滴水的方式一次显示每条消息,一条消息？

我msgget()在基于IPC的应用程序中使用函数.如何清理填充旧消息队列的队列？

我正在寻找在Ubuntu上运行的基于Web的系统的消息/队列解决方案。

该系统基于以下技术构建：

• Javascript（Extjs框架）-前端
• 的PHP
• Python（与加密设备交互的守护程序服务）
• Python pyserial-（串行端口交互）
• 的MySQL
• Linux-Ccustom bash脚本（更新数据库/邮件报告）

该系统具有以下目的：

• 在分布式平台上捕获客户端信息
• 使用硬件设备加密/解密敏感交易

系统故障：

1. 用户可以使用Web浏览器访问系统
2. 用户捕获客户端信息，然后按“提交”按钮数据被发送到加密设备，系统进入等待状态
3. 然后，数据在设备上加密并发送回浏览器
4. 加密的数据保存到数据库
5. 系统退出等待状态并显示DONE消息

请注意：我已经处理了等待/处理中的消息，所以忽略它。

到目前为止，我所做的是：

• 我创建了一个python守护程序，用于监视数据库视图中是否有任何新请求
• 守护程序服务使用pyserial在设备上执行新请求，并使用“响应”更新请求表。加密内容
• 我在PHP中创建了一个轮询服务，该服务经常检查>请求表中是否存在针对特定请求的“响应”
• 使用适当的等待/完成状态消息创建了Extjs前端

当前设置的问题：

• 并发-我们期望使用数据库作为消息/排队解决方案的任何时候，随时有20个以上的用户提交数据库的加密/解密请求，由于表锁定和仅1个监听请求的监听进程，因此无法扩展
• 守护程序服务-依赖守护程序服务有点冒险，并且数据库开销似乎有点高，每秒轮询一次视图以查找新请求
• 开发-通过仅将请求发送到加密/解密服务而不是执行在db中插入请求，轮询响应以及在守护程序服务中处理请求的整个过程，可以简化我的开发任务。

我的问题：

在这种情况下，理想的消息/查询解决方案是什么？请考虑>我的系统专门在Ubuntu O / S上运行。

我已经做了一些Google服务，遇到了一个叫做“ Stomp”的服务器，但是事实证明它安装起来有些困难，而且缺少一些文档。我也更喜欢那些在设置类似内容方面有经验的个人的建议，而不是一些“如何”指导的建议：)

感谢您的时间

我被告知有一段时间LMAX干扰器以及与标准消息队列相比的性能如何.我下载了最新版本,看到它是一个普通的JAR,包含许多类和类型,都围绕着它的超快速RingerBuffer对象.

最终,在一天结束时,基于队列的JMS提供程序将归结为管理Java队列对象(或者更可能是并发队列)的大量代码.所以在这方面,我看到了LMAX Disruptor和JMS提供者之间的比较(或者更确切地说,它是内部队列).

但是JMS提供者不仅仅是几个队列.它是一个完整的中间件应用程序,用于处理来自/来自消费者和生产者的消息.我想知道在LMAX土地上是否有JMS提供商？

以与任何其他JMS代理类似的方式连接到"Disruptor Broker"并向/从其读取/写入消息将是很好的.

这样的事情是存在的,还是我离开这里？

当我需要编写简单的时间服务器和使用Linux消息队列的客户端时,我有分配.服务器打开一个消息队列,客户端发送一个带有PID的消息(类型为1的消息),服务器读取该消息并发送一个PID类型的消息(取自读取的消息).我把所有代码都放在下面,因为我不知道我在哪里弄错了.我不是Linux编程专家.甚至不知道我写的服务器是否正确.

服务器和客户端包含的文件(我需要以这种方式编写).

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <sys/msg.h>
#include <sys/ipc.h>
#include <sys/types.h>
#include <string.h>
#include <signal.h>

#define QUEUE 100
#define PERM_ALL 0777

typedef struct my_msgbuf {
    long mtype;
    int pid;
} ClientMessage;

typedef struct my_msgbuf2 {
    long mtype;
    struct tm time;
} TimeMessage;

服务器

int m_queue;

void cleanup(int signum) {
    if (msgctl(m_queue, IPC_RMID, NULL) == -1) {
        printf("Something happen on clean up\n");
        exit(1);
    }
    exit(signum);
}

int main() {
    ClientMessage pid_message;
    TimeMessage t;
    time_t clock;
    struct tm *cur_time; …

以下是队列的理想设计:

• P制作人.插入数据的应用程序
• X交换.
• C1-C3消费者.从队列中读取的应用程序

队列详细信息:

• A.就像队列日志一样,如果没有客户端绑定,那么将丢弃消息.
• B.这是一个工作队列.如果有标准匹配,它会做一些事情.
• C.也是一个工作队列.它将改变数据

A是可选的,但BC将始终排队,直到某个客户端进程连接它.

问题是确定我应该使用哪种类型的交换.它是一个扇动,直接或主题？因为如果没有连接客户端,我希望A队列丢弃该消息,但B&C应始终保留该消息.

并且生产者应该一次写入交换,还是用不同的路由密钥或主题写多次？

我最近一直在阅读有关消息传递系统的内容,并专门研究了RabbitMQ和NServiceBus.正如我所理解的那样,如果某条消息由于某种原因而失败,则会再次尝试一次.然后,两个系统都提供了稍后再试的可能性,例如在5秒内.当五秒钟过去后,消息将再次发送多次.

我引用Vaughn Vernon 实现域驱动设计(p.502):

处理此问题的另一种方法是简单地重试发送直到成功,可能使用上限指数退避.在RabbitMQ的情况下,重试可能会失败很长一段时间.因此,使用消息NAK和重试的组合可能是最好的方法.尽管如此,如果我们的流程每五分钟重试三次,那么它可能就是我们所需要的.

对于NServiceBus,这称为二级重试,当重试发生时,它会多次发生.

为什么需要多次发生？为什么不每五分钟重试一次？五分钟后第一次重试失败的可能性是多少,第二次重试(可能是几毫秒后)应该成功的几率是多少？

如果它不需要由于某些配置(是吗？),为什么我发现的所有示例都有多次重试？

我想用Perl实现一个消息队列。我从stdin获取数据并将其发送到队列。

我邮件的结构是

struct message => {
    mtype        => '$',
    buffer_size  => '$',
    last_message => '$',
    buff         => '$',
};

我必须使用C程序从队列中接收数据。我的C程序以前运行良好，但是现在当我运行它以从队列中接收数据时，它显示出类似以下内容

age=HASH(0x1daa088) 1936942445      4000        

我使用在stdout上打印的4000字节缓冲区读取数据块。但是age=HASH(0x1daa088) 1936942445程序应该打印收到的消息的大小，而不是打印出来。

这里发生了什么？是因为C中的消息是结构，而Perl中的消息是哈希吗？

我的C代码：

age=HASH(0x1daa088) 1936942445      4000        

我的Perl代码：

#include <stdio.h>
#include <stdlib.h>
#include <linux/ipc.h>
#include <linux/msg.h>
#include <time.h>
#include <string.h>

#define bufsize 4000

struct mymsgbuf {
    long    mtype;          /* Message type */
    int     buffer_size;
    char    buff[bufsize];
    int     last_message;
} msg;

int read_message( int qid, long type, struct mymsgbuf *qbuf ) {

    int result, length;

    /* …

我的问题与这个问题有一些相似之处: 为什么像PostgreSQL这样的数据库需要像RabbitMQ这样的消息代理？

在我目前的(半专业)项目中,我也决定是否要使用数据库,基于消息代理(例如使用RabbitMQ),甚至是完全不同的解决方案.

让我们想象两个工具,工具A和工具B.每当工具A运行并完成时,工具B 可能有一些事情要做.工具A的执行需要安静一段时间(> 60秒)并且通常无需执行任何操作工具B.工具A为工具B提供了一些元数据,因此工具B知道该怎么做.

基于消息的解决方案:建立工具B正在使用的消息队列.如果执行工具A 并且工具B应该运行,则工具A将消息(包括元数据)发布到工具B接收的队列,以便工具B将使用消息中的元数据运行.

数据库解决方案:每当工具A运行时,它都会添加一个数据库记录,例如时间戳,元数据和状态"RUNNING".如果执行工具A 并且工具B应该运行,它会将DB记录状态更新为"NEXT_TOOL_B".工具B不断向DB查询"NEXT_TOOL_B"状态的记录.如果找到某些内容,工具B将使用DB记录中的元数据运行.

虽然我知道数据库解决方案的缺点,例如工具B的常量轮询,但我在基于消息的解决方案中遗漏了它的一个特性:

每当第三个工具(例如工具C,例如控制面板UI)想知道当前状态时,它也可以随时查询数据库,并且如果工具A仍在工作,它将发现"运行"状态.在消息解决方案中,我并没有真正看到"监视"状态的方法,除非完成消息将在队列中.

所以我的问题是,你能想到使用消息或任何其他方法实现这一点而不进行轮询吗？

lparallel库中队列的基本讨论，网址为https://z0ltan.wordpress.com/2016/09/09/basic-concurrency-and-parallelism-in-common-lisp-part-4a-parallelism-using-lparallel- basics /＃channels说，该队列“启用在工作线程之间传递消息”。下面的测试使用一个共享队列来协调主线程和从属线程，其中主线程在退出之前只是等待下一个线程的完成：

(defun foo (q)
  (sleep 1)
  (lparallel.queue:pop-queue q))  ;q is now empty

(defun test ()
  (setf lparallel:*kernel* (lparallel:make-kernel 1))
  (let ((c (lparallel:make-channel))
        (q (lparallel.queue:make-queue)))
    (lparallel.queue:push-queue 0 q)
    (lparallel:submit-task c #'foo q)
    (loop do (sleep .2)
             (print (lparallel.queue:peek-queue q))
          when (lparallel.queue:queue-empty-p q)
            do (return)))
  (lparallel:end-kernel :wait t))

这可以按预期的方式产生输出：

* (test)

0
0
0
0
NIL
(#<SB-THREAD:THREAD "lparallel" FINISHED values: NIL {10068F2B03}>)

我的问题是我是否正确或完全使用了lparallel的队列功能。似乎队列只是使用全局变量来保存线程共享对象的替代品。使用队列的设计优势是什么？通常为每个提交的任务分配一个队列（假设任务需要进行通信）是一种好习惯吗？感谢您提供更深入的见解。