BeansTalkd 做消息队列服务

无意间看到这个仓库讲php关于 BeanStalkd 的扩展，然后就去了解了一下beanstalkd，用它可以用来做队列服务。

话不多说，安装一下试试。

首先 sudo apt search beanstalk 搜索一下发现


Sorting... Done
Full Text Search... Done
awscli/focal-updates,focal-updates 1.18.69-1ubuntu0.20.04.1 all
  Universal Command Line Environment for AWS

beanstalkd/focal,now 1.11-1 amd64 [installed]
  simple, in-memory, workqueue service

python-celery-common/focal,focal 4.2.1-5ubuntu1 all
  async task/job queue based on message passing (common files)

python-celery-doc/focal,focal 4.2.1-5ubuntu1 all
  async task/job queue based on message passing (Documentation)

python3-celery/focal,focal 4.2.1-5ubuntu1 all
  async task/job queue based on message passing (Python3 version)

ruby-beaneater/focal,focal 1.0.0-1 all
  simple beanstalkd client for Ruby
 

好了，找到这个beanstalkd了

然后安装

 sudo apt-get install beanstalkd 

Reading package lists... Done
Building dependency tree       
Reading state information... Done
Suggested packages:
  doc-base
The following NEW packages will be installed:
  beanstalkd
0 upgraded, 1 newly installed, 0 to remove and 90 not upgraded.
Need to get 43.9 kB of archives.
After this operation, 125 kB of additional disk space will be used.
Get:1 http://cn.archive.ubuntu.com/ubuntu focal/universe amd64 beanstalkd amd64 1.11-1 [43.9 kB]
Get:1 http://cn.archive.ubuntu.com/ubuntu focal/universe amd64 beanstalkd amd64 1.11-1 [43.9 kB]
Fetched 31.3 kB in 4s (7,047 B/s)     
Selecting previously unselected package beanstalkd.
(Reading database ... 256731 files and directories currently installed.)
Preparing to unpack .../beanstalkd_1.11-1_amd64.deb ...
Unpacking beanstalkd (1.11-1) ...
Setting up beanstalkd (1.11-1) ...
Created symlink /etc/systemd/system/multi-user.target.wants/beanstalkd.service → /lib/systemd/system/beanstalkd.service.
beanstalkd.socket is a disabled or a static unit, not starting it.
Processing triggers for man-db (2.9.1-1) ...
Processing triggers for systemd (245.4-4ubuntu3.11) ...
 

安装成功，这个东西会监听 11300 端口
然后使用这个工具来看看


监控工具：

wget https://github.com/src-d/beanstool/releases/download/v0.2.0/beanstool_v0.2.0_linux_amd64.tar.gz

获取文件后解压

tar -xvzf beanstool_v0.2.0_linux_amd64.tar.gz

然后拷贝到 /usr/local/bin/

sudo cp beanstool_v0.2.0_linux_amd64/beanstool /usr/local/bin/

这样就直接用 beanstool 了

查看当前状态

beanstool stats

结果

+---------+----------+----------+----------+----------+----------+----------+----------+
| Name    | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+---------+----------+----------+----------+----------+----------+----------+----------+
| default | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
+---------+----------+----------+----------+----------+----------+----------+----------+

然后使用composer的vendor包

composer require pda/pheanstalk
安装完毕后创建一个input.php文件做生产者


<?php
require __DIR__ . '/vendor/autoload.php';

use Pheanstalk\Pheanstalk;

$pheanstalk = Pheanstalk::create('127.0.0.1');

// Queue a Job
$pheanstalk
  ->useTube('testtube')
  ->put("job payload goes here\n");

$pheanstalk
    ->useTube('testtube')
    ->put(
        json_encode(['test' => 'data']),  // encode data in payload
        Pheanstalk::DEFAULT_PRIORITY,     // default priority
        30, // delay by 30s
        60  // beanstalk will retry job after 60s
     );


再创建一个output.php文件做消费者


<?php
require __DIR__ . '/vendor/autoload.php';
use Pheanstalk\Pheanstalk;

$pheanstalk = Pheanstalk::create('127.0.0.1');

// we want jobs from 'testtube' only.
$pheanstalk->watch('testtube');

// this hangs until a Job is produced.
$job = $pheanstalk->reserve();

try {
    $jobPayload = $job->getData();
    // do work.

    sleep(2);
    // If it's going to take a long time, periodically
    // tell beanstalk we're alive to stop it rescheduling the job.
    $pheanstalk->touch($job);
    sleep(2);

    // eventually we're done, delete job.
    $pheanstalk->delete($job);
}
catch(\Exception $e) {
    // handle exception.
    // and let some other worker retry.
    $pheanstalk->release($job); 
}
 

然后执行一下  php input.php

查看 状态

 
$ beanstool stats
+----------+----------+----------+----------+----------+----------+----------+----------+
| Name     | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+----------+----------+----------+----------+----------+----------+----------+----------+
| default  | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
| testtube | 0        | 1        | 1        | 0        | 0        | 0        | 2        |
+----------+----------+----------+----------+----------+----------+----------+----------+
 
我们看到有一个在ready状态，一个在delayed状态，这是由于第二次的put采用了延时30s，然后过一段时间后再看

 
$ beanstool stats
+----------+----------+----------+----------+----------+----------+----------+----------+
| Name     | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+----------+----------+----------+----------+----------+----------+----------+----------+
| default  | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
| testtube | 0        | 0        | 2        | 0        | 0        | 0        | 2        |
+----------+----------+----------+----------+----------+----------+----------+----------+
复制代码
已经有2个在ready状态了。

此时我们用消费者执行一下 php output.php 

与此同时迅速看状态

复制代码
$ beanstool stats
+----------+----------+----------+----------+----------+----------+----------+----------+
| Name     | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+----------+----------+----------+----------+----------+----------+----------+----------+
| default  | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
| testtube | 0        | 0        | 1        | 1        | 0        | 0        | 2        |
+----------+----------+----------+----------+----------+----------+----------+----------+
再次执行
$ beanstool stats
+----------+----------+----------+----------+----------+----------+----------+----------+
| Name     | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+----------+----------+----------+----------+----------+----------+----------+----------+
| default  | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
| testtube | 0        | 0        | 1        | 0        | 0        | 0        | 2        |
+----------+----------+----------+----------+----------+----------+----------+----------+
复制代码
因为我们有sleep(2)，所以要尽量快点操作这个状态监控的命令，可以看到有一个拿出来放入了reserved，然后就消失了(实际上这是后面的代码delete导致的，因为已经消费完毕)

再次执行 php output.php 

 
$ beanstool stats:
+----------+----------+----------+----------+----------+----------+----------+----------+
| Name     | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+----------+----------+----------+----------+----------+----------+----------+----------+
| default  | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
| testtube | 0        | 0        | 0        | 1        | 0        | 0        | 2        |
+----------+----------+----------+----------+----------+----------+----------+----------+

$ beanstool stats:
+---------+----------+----------+----------+----------+----------+----------+----------+
| Name    | Buried   | Delayed  | Ready    | Reserved | Urgent   | Waiting  | Total    |
+---------+----------+----------+----------+----------+----------+----------+----------+
| default | 0        | 0        | 0        | 0        | 0        | 0        | 0        |
+---------+----------+----------+----------+----------+----------+----------+----------+
 
同样也是迅速观测这个状态，发现消费1个，然后删除1个，现在队列空了，这说明确实是符合我们的期望的。

然后回到文章开头提到的扩展，这个扩展就是帮我们实现了composer装的那个pheanstalk包。


这个扩展如何安装呢？

步骤如下：

克隆项目

git clone https://gitee.com/qzfzz/php-beanstalk.git
进行编译 

phpize

然后

./configure

之后

make && make install

sudo make install

然后修改 php.ini

sudo gedit /etc/php/7.4/cli/php.ini /etc/php/7.4/apache2/php.ini

加上这句

extension=beanstalk

重启apache2

sudo /etc/init.d/apache2 restart
之后就可以使用这个扩展了。

这个扩展它封装为函数了，可以看到他有个例子文件

简单的找了几个例子

复制代码
<?php

$config = [
  'host' => '127.0.0.1',
  'port' => 11300
];
$beanstalk_obj = beanstalk_connect($config['host'], $config['port']);
$last_job_id = beanstalk_put($beanstalk_obj, "message detail");
beanstalk_delete($beanstalk_obj, $last_job_id);
$last_job_id = beanstalk_putInTube($beanstalk_obj, 'tubea', "message detail");
复制代码
可以看到使用 connect 连接， put 塞入新的job消息， putInTube 来塞入指定管道的tubea，delete来删除等等，具体可以看看源代码学习一下，我对比了一下这两种方式实现效率。

第一种采用composer包(我还特意去掉了加载文件所需要的时间)

复制代码
<?phprequire __DIR__ . '/vendor/autoload.php';

use Pheanstalk\Pheanstalk;

$start = microtime( true );
$pheanstalk = Pheanstalk::create('127.0.0.1');

$pheanstalk
  ->useTube('testtube')
  ->put(date("Y-m-d H:i:s") . "job payload goes here\n");

$end = microtime(true);

echo ($end - $start) * 1000 . " ms";
复制代码
执行需要2.59ms

第二种直接用扩展函数


<?php
$start = microtime(true);

$config = [
  'host' => '127.0.0.1',
  'port' => 11300
];

$beanstalk_object = beanstalk_connect($config['host'], $config['port']);
$last_job_id = beanstalk_putInTube($beanstalk_object, 'testtube', date("Y-m-d H:i:s") . "job payload goes here\n");

$end = microtime(true);

echo ($end - $start) * 1000 . " ms";
复制代码
执行需要0.34ms

不得不说，扩展就是扩展，就是快的多啊！

我另外测试了一下投递极限

循环投递10000次消息，大概在500ms左右

复制代码
$start = microtime( true );
for ($i=0; $i < 10000; $i++) { 
  $pheanstalk
  ->useTube('testtube')
  ->put(date("Y-m-d H:i:s") . "job payload goes here\n");
}
$end = microtime( true );
echo ($end - $start) * 1000 . " ms";

 
也就意味着1秒钟只能投递20000条消息，这比rabbitmq的投递慢多了(参见这篇文章)

但是它执行1次投递消息的时候却比这个rabbitmq的代码执行的快，原因是我测试了一下这个rabbitmq的连接上就耗费了9ms

$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');

更别提还有这两步了

$channel = $connection->channel();
$channel->queue_declare('hello', false, false, false, false);
这样来看，想迅速投递一条短消息，或者建立小信息量的job用beanstalk是很不错的，如果有大量消息集中投递，那使用 rabbitmq 是很不错的。

另外这个beanstalk投递可以延时，非常适合有些时候需要在当前时间一段时间后执行某个任务，往后弄个类似于一次性的定时器的功能，这个东西值得尝试。

而且如果使用while死循环将 output.php 脚本一直挂着跑，它就能一直消费消息了，这样就等于有个后端进程一直能帮我们做消费者处理堆积的任务了，特殊场景可以考虑一下这个方案。

Windows 安装 Beanstalkd

下载并安装 cygwin

1. 添加镜像地址 - http://mirrors.aliyun.com
2. 下载 `gcc`、`gcc-core`、`make`、`automake`, 在 Devel 分支下

下载 beanstalkd-win 包

1. 解压并进入beanstalkd-win目录
2. 打开CMD窗口,运行 ./beanstalkd.exe -l 127.0.0.1 -p 11300
 

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/152442.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

C练习题_14

C练习题_14

一、单项选择题&#xff08;本大题共 20小题&#xff0c;每小题 2分&#xff0c;共 40分。在每小题给出的四个备选项中&#xff0c;选出一个正确的答案&#xff0c;并将所选项前的字母填写在答题纸的相应位置上。) 以下叙述不正确的是&#xff08;&#xff09; A.一个C源程序可…
阅读更多...
高版本Vivado和Linux 4.x内核移植Digilent Driver

高版本Vivado和Linux 4.x内核移植Digilent Driver

移植环境 Vivado 2022.2Ubuntu 22.04petalinux 2022.2Linux内核4.14&#xff08;xilinx-linux-2018.3&#xff09;linux-digilent 主要问题 https://github.com/Digilent/linux-digilent 这些驱动支持Linux kernel release 4.x&#xff0c;然而和Vitis 2022.2 套件对应的内核…
阅读更多...
buildadmin+tp8表格操作（8） 表格下方添加 合计行

buildadmin+tp8表格操作（8） 表格下方添加 合计行

表格的下方可以自定义添加一个合计行&#xff0c;如果有其它的需求&#xff0c; 我们可以添加我们自已需要的行&#xff0c; 并不局限于合计行 以上就可以给表格的最下方添加一个合计行了 完整代码如下 <template><div class"default-main ba-table-box"&…
阅读更多...
C语言十六弹 --求两个整数二进制位不同的位数

C语言十六弹 --求两个整数二进制位不同的位数

求两个整数二进制位不同的位数 思路&#xff1a;1.要求不同的个数 就必须遍历比较两个数的二进制位&#xff0c;不同就使用一个三方变量接收&#xff0c;相同则跳过。 2.使用一个相同的三方变量来作为两者判断条件的基础&#xff0c;而考虑到需要遍历二进制位 则使用1来作为三…
阅读更多...
AVL树你需要了解一下

AVL树你需要了解一下

AVL树介绍 AVL树是一种自平衡二叉查找树&#xff0c;它得名于发明者G.M.Adel’son-Vel’skii和E.M.Landis。AVL树的特点是任何节点的两个子树的高度最大差别为1&#xff0c;因此它也被称为高度平衡树。在AVL树中&#xff0c;每个节点的平衡因子只有-1、0、1三种&#xff0c;通…
阅读更多...
人工智能给我们的生活带来了巨大的影响？

人工智能给我们的生活带来了巨大的影响？

1. 人工智能从哪些方面给我们带来了影响&#xff1f; 人工智能出现&#xff0c;极大地影响了人类的生活&#xff0c;下面是人工智能所影响的领域&#xff1a; 1. 日常生活 智能家居: AI驱动的设备&#xff0c;如智能扬声器、灯光、恒温器&#xff0c;正在改变我们与家居环境的…
阅读更多...
【鸿蒙最新全套教程】＜HarmonyOS第一课＞1、运行Hello World

【鸿蒙最新全套教程】＜HarmonyOS第一课＞1、运行Hello World

下载与安装DevEco Studio 在HarmonyOS应用开发学习之前&#xff0c;需要进行一些准备工作&#xff0c;首先需要完成开发工具DevEco Studio的下载与安装以及环境配置。 进入DevEco Studio下载官网&#xff0c;单击“立即下载”进入下载页面。 DevEco Studio提供了Windows版本和…
阅读更多...
文件上传漏洞（CVE-2022-23043）

文件上传漏洞（CVE-2022-23043）

简介 CVE-2022-23043是一个与Zenario CMS 9.2文件上传漏洞相关的安全漏洞。该漏洞被定义为文件的不加限制上传&#xff0c;攻击者可以利用这个漏洞上传webshell以执行任意命令。利用这个漏洞的攻击者暂无特定情况。要利用此漏洞&#xff0c;攻击者首先需要访问Zenario CMS的管…
阅读更多...
Django 路由配置（二）

Django 路由配置（二）

一、路由 就是根据用户请求的URL链接来判断对应的出来程序&#xff0c;并返回处理结果&#xff0c;也是就是URL和django的视图建立映射关系. 二、Django请求页面的步骤 1、首先Django确定要使用的根URLconf模块&#xff0c;通过ROOT_URLCONF来设置&#xff0c;在settings.py配置…
阅读更多...
服务器数据恢复—OCFS2下raid5磁盘损坏导致阵列崩溃的数据恢复案例

服务器数据恢复—OCFS2下raid5磁盘损坏导致阵列崩溃的数据恢复案例

服务器数据恢复环境&#xff1a; IBM某型号存储&#xff0c;6块sas硬盘组建一组raid5&#xff0c;划分一个lun分配给Linux服务器并格式化为OCFS2文件系统&#xff0c;共享给虚拟化使用&#xff0c;存放的数据包括24台liunx和windows虚拟机、压缩包文件和配置文件。 服务器故障…
阅读更多...
学习笔记6——垃圾回收

学习笔记6——垃圾回收

学习笔记系列开头惯例发布一些寻亲消息 链接&#xff1a;https://baobeihuijia.com/bbhj/contents/3/190801.html java垃圾回收&#xff08;stop the world&#xff09; 专注于堆和方法区的垃圾回收&#xff0c;年轻代&#xff0c;老年代&#xff0c;永久代判断对象是否还存…
阅读更多...
Java8Stream快速使用

Java8Stream快速使用

将List集合存入流中 List<String> list new ArrayList<>();list.add("张一");list.add("张二");list.add("张三");list.add("李四");list.add("赵五");list.add("张六");list.add("王八"…
阅读更多...
Linux进程通信——IPC、管道、FIFO的引入

Linux进程通信——IPC、管道、FIFO的引入

进程间的通信——IPC 进程间通信 (IPC&#xff0c;InterProcess Communication) 是指在不同进程之间传播或交换信息。 IPC的方式通常有管道 (包括无名管道和命名管道) 、消息队列、信号量、共享存储、Socket、Streams等。其中 Socket和Streams支持不同主机上的两个进程IPC。 …
阅读更多...
Spring的声明式事务

Spring的声明式事务

1. Spring声明式事务的使用流程 在Spring中使用声明式事务主要涉及以下几个步骤&#xff1a; 使用 DataSourceTransactionManager 是针对基于JDBC的应用程序。以下是在Spring框架中使用 DataSourceTransactionManager 的步骤&#xff1a; 1. 引入相关依赖 确保你的项目中包…
阅读更多...
pyQt主界面与子界面切换简易框架

pyQt主界面与子界面切换简易框架

本篇来介绍使用python中是Qt功能包&#xff0c;设置一个简易的多界面切换框架&#xff0c;实现主界面和多个子界面直接的切换显示。 1 主界面 设计的Demo主界面如下&#xff0c;主界面上有两个按钮图标&#xff0c;点击即可切换到对应的功能界面中&#xff0c;进入子界面后&a…
阅读更多...
【mediasoup】TransportCongestionControlClient 1： 代码走读

【mediasoup】TransportCongestionControlClient 1： 代码走读

TransportCongestionControlClient 基于m77版本的libwebrtc ，但是TransportCongestionControlClient 并不是libwebrt中的，是mediasoup自己封装实现：TransportCongestionControlClient 用于发送端D:\XTRANS\soup\mediasoup-sfu-cpp\worker\src\RTC\TransportCongestionContro…
阅读更多...
（动手学习深度学习）第13章 实战kaggle竞赛：CIFAR-10

（动手学习深度学习）第13章 实战kaggle竞赛：CIFAR-10

导入相关库 import collections import math import os import shutil import pandas as pd import torch import torchvision from torch import nn from d2l import torch as d2l下载数据集 d2l.DATA_HUB[cifar10_tiny] (d2l.DATA_URL kaggle_cifar10_tiny.zip,2068874e4…
阅读更多...
CentOS 8搭建WordPress

CentOS 8搭建WordPress

步骤 1: 更新系统 确保你的系统是最新的&#xff0c;使用以下命令更新&#xff1a; bashCopy code sudo dnf update 步骤 2: 安装Apache bashCopy code sudo dnf install httpd 启动Apache&#xff0c;并设置开机自启动&#xff1a; bashCopy code sudo systemctl star…
阅读更多...
C语言--每日五道选择题--Day20

C语言--每日五道选择题--Day20

第一题 1. 在如下结构定义中&#xff0c;不正确的是&#xff08; &#xff09;。 A&#xff1a; struct student { 　int no; 　char name[10]; 　float score; }; B&#xff1a; struct stud[20] { 　int no; 　char name[10]; 　float score; }; C&#xff1a; struct stu…
阅读更多...
ElasticSearch在Windows上的下载与安装

ElasticSearch在Windows上的下载与安装

Elasticsearch是一个开源的分布式搜索和分析引擎&#xff0c;它可以帮助我们快速地搜索、分析和处理大量数据。Elasticsearch能够快速地处理结构化和非结构化数据&#xff0c;支持全文检索、地理位置搜索、自动补全、聚合分析等功能&#xff0c;能够承载各种类型的应用&#xf…
阅读更多...
最新文章

Copyright @ 2022~2023