基于 Redis 的订阅与发布

github 仓库

demo-redis-subscribe

创建项目

$ compor create hyperf/biz-skeleton demo-redis-subscribe dev-masterinstalling hyperf/biz-skeleton (dev-master 1eaa35a957cf704a5c9959c68e426a614c7598a2)- installing hyperf/biz-skeleton (dev-master 1eaa35a): cloning 1eaa35a957 from cachecreated project in demo-redis-subscribe> @php -r "file_exists('.env') || copy('.env.example', '.env');"loading compor repositories with package informationupdating dependencies (including require-dev)package operations: 133 installs, 0 updates, 0 removals- installing ocramius/package-versions (1.4.0): loading from cache- installing hyperf/contract (dev-master 1624d1c): cloning 1624d1ce0e from cache- installing doctrine/inflector (v1.3.0): loading from cache- installing hyperf/utils (dev-master c847116): cloning c847116cf9 from cache- installing psr/container (1.0.0): loading from cache- installing hyperf/pool (dev-master 5a67570): cloning 5a67570f71 from cache- installing psr/event-dispatcher (1.0.0): loading from cache- installing hyperf/process (dev-master 75f4fbf): cloning 75f4fbfa56 from cache- installing doctrine/instantiator (1.2.0): loading from cache- installing psr/log (1.1.0): loading from cache- installing php-amqplib/php-amqplib (v2.10.0): loading from cache- installing hyperf/amqp (dev-master 1cff378): cloning 1cff378bc4 from cache- installing symfony/rvice-contracts (v1.1.6): loading from cache- installing symfony/polyfill-php73 (v1.12.0): loading from cache- installing symfony/polyfill-mbstring (v1.12.0): loading from cache- installing symfony/console (v4.3.4): loading from cache- installing hyperf/command (dev-master e4caea4): cloning e4caea4c70 from cache- installing hyperf/async-queue (dev-master 8d28229): cloning 8d282297cd from cache- installing psr/simple-cache (1.0.1): loading from cache- installing hyperf/cache (dev-master 60997fb): cloning 60997fb05c from cache- installing hyperf/circuit-breaker (dev-master e41726f): cloning e41726fd25 from cache- installing symfony/finder (v4.3.4): loading from cache- installing symfony/polyfill-ctype (v1.12.0): loading from cache- installing phpoption/phpoption (1.5.0): loading from cache- installing vlucas/phpdotenv (v3.6.0): loading from cache- installing hyperf/config (dev-master 44f5ef8): cloning 44f5ef8a10 from cache- installing psr/http-message (1.0.1): loading from cache- installing fig/http-message-util (1.1.3): loading from cache- installing hyperf/framework (dev-master 3be169b): cloning 3be169b9f4 from cache- installing hyperf/event (dev-master e9d97f1): cloning e9d97f11fb from cache- installing php-di/phpdoc-reader (2.1.0): load三国演义人物赏析ing from cache- installing doctrine/lexer (1.1.0): loading from cache- installing doctrine/annotations (v1.7.0): loading from cache- installing nikic/php-parr (v4.2.4): loading from cache- installing hyperf/di (dev-master 7f82227): cloning 7f822276a0 from cache- installing hyperf/constants (dev-master a3baaf8): cloning a3baaf8bbd from cache- installing hyperf/devtool (dev-master c3f424f): cloning c3f424f2c6 from cache- installing ralouphie/getallheaders (3.0.3): loading from cache- installing guzzlehttp/psr7 (1.6.1): loading from cache- installing guzzlehttp/promis (v1.3.1): loading from cache- installing guzzlehttp/guzzle (6.3.3): loading from cache- installing hyperf/guzzle (dev-master 90ad6e4): cloning 90ad6e497b from cache- installing react/promi (v2.7.1): loading from cache- installing guzzlehttp/streams (3.0.0): loading from cache- installing guzzlehttp/ringphp (1.1.1): loading from cache- installing elasticarch/elasticarch (v6.7.2): loading from cache- installing hyperf/elasticarch (dev-master c936fc0): cloning c936fc0f24 from cache- installing monolog/monolog (1.25.1): loading from cache- installing hyperf/logger (dev-master 937101c): cloning 937101c682 from cache- installing hyperf/memory (dev-master 17003ec): cloning 17003ec1e9 from cache- installing hyperf/paginator (dev-master 4ffeb0e): cloning 4ffeb0e46c from cache- installing symfony/translation-contracts (v1.1.6): loading from cache- installing symfony/translation (v4.3.4): loading from cache- installing nesbot/carbon (2.24.0): loading from cache- installing hyperf/databa (dev-master d36785e): cloning d36785e709 from cache- installing hyperf/model-listener (dev-master 5a77c83): cloning 5a77c83cbe from cache- installing hyperf/db-connection (dev-master 17eec65): cloning 17eec65d60 from cache- installing hyperf/model-cache (dev-master e3c5293): cloning e3c5293a56 from cache- installing hyperf/redis (dev-master 6043083): cloning 6043083650 from cache- installing hyperf/rver (dev-master b0东岳泰山的资料c8bd9): cloning b0c8bd9e92 from cache- installing zendframework/zend-stdlib (3.2.1): loading from cache- installing zendframework/zend-mime (2.7.1): loading from cache- installing hyperf/http-message (dev-master dcd1d78): cloning dcd1d78b88 from cache- installing psr/http-rver-handler (1.0.1): loading from cache- installing psr/http-rver-middleware (1.0.1): loading from cache- installing hyperf/dispatcher (dev-master 670f7dc): cloning 670f7dca4f from cache- installing hyperf/exception-handler (dev-master d1f3511): cloning d1f3511c9c from cache- installing nikic/fast-route (v1.3.0): loading from cache- installing hyperf/http-rver (dev-master 2744b1a): cloning 2744b1a949 from cache- installing phpunit/php-timer (2.1.2): loading from cache- installing bastian/environment (4.2.2): loading from cache- installing bastian/version (2.0.1): loading from cache- installing bastian/resource-operations (2.0.1): loading from cache- installing bastian/object-reflector (1.1.1): loading from cache- installing bastian/recursion-context (3.0.0): loading from cache- installing bastian/object-enumerator (3.0.3): loading from cache- installing bastian/global-state (2.0.0): loading from cache- installing bastian/exporter (3.1.2): loading from cache- installing bastian/diff (3.0.2): loading from cache- installing bastian/comparator (3.0.2): loading from cache- installing phpunit/php-text-template (1.2.1): loading from cache- installing phpunit/php-file-iterator (2.0.2): loading from cache- installing theer/tokenizer (1.1.3): loading from cache- installing bastian/code-unit-rever-lookup (1.0.1): loading from cache- installing phpunit/php-token-stream (3.1.1): loading from cache- installing phpunit/php-code-coverage (6.1.4): loading from cache- installing webmozart/asrt (1.5.0): loading from cache- installing phpdocumentor/reflection-common (2.0.0): loading from cache- installing phpdocumentor/type-resolver (1.0.1): loading from cache- installing phpdocumentor/reflection-docblock (4.3.2): loading from cache- installing phpspec/prophecy (1.8.1): loading from cache- installing phar-io/version (2.0.1): loading from cache- installing phar-io/manifest (1.0.3): loading from cache- installing myclabs/deep-copy (1.9.3): loading from cache- installing phpunit/phpunit (7.5.16): loading from cache- installing hyperf/testing (dev-master 910f9e9): cloning 910f9e94db from cache- installing swoft/swoole-ide-helper (v4.4.6): loading from cache- installing symfony/filesystem (v4.3.4): loading from cache- installing symfony/config (v4.3.4): loading from cache- installing symfony/dependency-injection (v4.3.4): loading from cache- installing pdepend/pdepend (2.5.2): loading from cache- installing phpmd/phpmd (2.7.0): loading from cache- installing symfony/stopwatch (v4.3.4): loading from cache- installing symfony/process (v4.3.4): loading from cache- installing symfony/polyfill-php72 (v1.12.0): loading from cache- installing paragonie/random_compat (v9.99.99): loading from cache- installing symfony/polyfill-php70 (v1.12.0): loading from cache- installing symfony/options-resolver (v4.3.4): loading from cache- installing symfony/event-dispatcher-contracts (v1.1.5): loading from cache- installing symfony/event-dispatcher (v4.3.4): loading from cache- installing php-cs-fixer/diff (v1.3.0): loading from cache- installing compor/xdebug-handler (1.3.3): loading from cache- installing compor/mver (1.5.0): loading from cache- installing friendsofphp/php-cs-fixer (v2.15.3): loading from cache- installing hamcrest/hamcrest-php (v2.0.0): loading from cache- installing mockery/mockery (1.2.3): loading from cache- installing doctrine/reflection (v1.0.0): loading from cache- installing doctrine/event-manager (v1.0.0): loading from cache- installing doctrine/collections (v1.6.2): loading from cache- installing doctrine/cache (v1.8.0): loading from cache- installing doctrine/persistence (1.1.1): loading from cache- installing doctrine/common (v2.11.0): loading from cache- installing phpstan/phpdoc-parr (0.3.5): loading from cache- installing nette/utils (v3.0.1): loading from cache- installing nette/schema (v1.0.0): loading from cache- installing nette/finder (v2.5.1): loading from cache- installing nette/robot-loader (v3.2.0): loading from cache- installing nette/php-generator (v3.2.3): loading from cache- installing nette/neon (v3.0.0): loading from cache- installing nette/di (v3.0.1): loading from cache- installing nette/bootstrap (v3.0.0): loading from cache- installing jean85/pretty-package-versions (1.2): loading from cache- installing phpstan/phpstan (0.11.16): loading from cachehyperf/utils suggests installing symfony/var-dumper (required to u the dd function (^4.1).)hyperf/utils suggests installing symfony/rializer (required to u symfonynormalizer (^4.3))hyperf/utils suggests installing symfony/property-access (required to u symfonynormalizer (^4.3))symfony/console suggests installing symfony/lockguzzlehttp/psr7 suggests installing zendframework/zend-httphandlerrunner (emit psr-7 respons)monolog/monolog suggests installing graylog2/gelf我们大家跳起来-php (allow nding log messages to a graylog2 rver)monolog/monolog suggests installing ntry/ntry (allow nding log messages to a ntry rver)monolog/monolog suggests installing doctrine/couchdb (allow nding log messages to a couchdb rver)monolog/monolog suggests installing ruflin/elastica (allow nding log messages to an elastic arch rver)monolog/monolog suggests installing ext-amqp (allow nding log messages to an amqp rver (1.0+ required))monolog/monolog suggests installing ext-mongo (allow nding log messages to a mongodb rver)monolog/monolog suggests installing mongodb/mongodb (allow nding log messages to a mongodb rver via php driver)monolog/monolog suggests installing aws/aws-sdk-php (allow nding log messages to aws rvices like dynamodb)monolog/monolog suggests installing rollbar/rollbar (allow nding log messages to rollbar)monolog/monolog suggests installing php-console/php-console (allow nding log messages to google chrome)symfony/translation suggests installing symfony/yamlhyperf/dat十个严禁心得体会aba suggests installing doctrine/dbal (required to rename columns (^2.6).)zendframework/zend-mime suggests installing zendframework/zend-mail (zend\mail component)bastian/global-state suggests installing ext-uopz (*)phpunit/php-code-coverage suggests installing ext-xdebug (^2.6.0)phpunit/phpunit suggests installing phpunit/php-invoker (^2.0)phpunit/phpunit suggests installing ext-xdebug (*)symfony/config suggests installing symfony/yaml (to u the yaml reference dumper)symfony/dependency-injection suggests installing symfony/yamlsymfony/dependency-injection suggests installing symfony/expression-language (for using expressions in rvice container configuration)symfony/dependency-injection suggests installing symfony/proxy-manager-bridge (generate rvice proxies to lazy load them)paragonie/random_compat suggests installing ext-libsodium (provides a modern crypto api that can be ud to generate random bytes.)symfony/event-dispatcher suggests installing symfony/http-kernelfriendsofphp/php-cs-fixer suggests installing php-cs-fixer/phpunit-constraint-isidenticalstring (for isidenticalstring constraint.)friendsofphp/php-cs-fixer suggests installing php-cs-fixer/phpunit-constraint-xmlmatchesxsd (for xmlmatchesxsd constraint.)doctrine/cache suggests installing alcaeus/mongo-php-adapter (required to u legacy mongodb driver)nette/bootstrap suggests installing tracy/tracy (to u configurator::enabletracy())writing lock filegenerating autoload filesocramius/package-versions:  generating version class...ocramius/package-versions: ...done generating version classdo you want to remove the existing vcs (.git, .svn..) history? [y,n]? y

定义队列

我们希望可以监听多个队列，所以便把队列名字额外定义出来。

<?phpdeclare(strict_types=1);/*** this file is part of hyperf.** @link     https://www.hyperf.io* @document https://doc.hyperf.io* @contact  group@hyperf.io* @licen  https://github.com/hyperf-cloud/hyperf/blob/master/licen*/namespace app\constants;class channel{const queue = 'channel.queue';const test = 'channel.test';public static function getarray(){return [lf::queue,lf::test,];}}

创建用于订阅的 redis 实例

更新配置

autoload/redis.php中新增subscriber相关配置

<?phpdeclare(strict_types=1);/*** this file is part of hyperf.** @link     https://www.hyperf.io* @document https://doc.hyperf.io* @contact  group@hyperf.io* @licen  https://github.com/hyperf-cloud/hyperf/blob/master/licen*/return ['default' => ['host' => env('redis_host', 'localhost'),'auth' => env('redis_auth', null),'port' => (int) env('redis_port', 6379),'db' => (int) env('redis_db', 0),'pool' => ['min_connections' => 1,'max_connections' => 10,'connect_timeout' => 10.0,'wait_timeout' => 3.0,'heartbeat' => -1,'max_idle_time' => (float) env('redis_max_idle_time', 60),],],'subscriber' => ['host' => env('redis_host', 'localhost'),'auth' => env('redis_auth', null),'port' => (int) env('redis_port', 6379),'db' => (int) env('redis_db', 0),'options' => [\redis::opt_read_timeout => '-1',],'pool' => ['min_connections' => 1,'max_connections' => 10,'connect_timeout' => 10.0,'wait_timeout' => 3.0,'heartbeat' => -1,'max_idle_time' => (float) env('redis_max_idle_time', 60),],],];

增加subscriber客户端

<?php<?phpdeclare(strict_types=1);/*** this file is part of hyperf.** @link     https://www.hyperf.io* @document https://doc.hyperf.io* @contact  group@hyperf.io* @licen  https://github.com/hyperf-cloud/hyperf/blob/master/licen*/namespace app\process;u app\constants\channel;u app\kernel\subscriber;u hyperf\process\abstractprocess;u hyperf\process\annotation\process;/*** @process(name="redisconsumerprocess")*/class redisconsumerprocess extends abstractprocess{public function handle(): void{$redis = di()->get(subscriber::class);$redis->subscribe(channel::getarray(), function ($instance, $channelname, $message) {var_dump($instance);var_dump($channelname);var_dump($message);// todo: 执行对应的消费操作});}}

测试代码

修改indexcontroller如下

<?phpdeclare(strict_types=1);/*** this file is part of hyperf.** @link     https://www.hyperf.io* @document https://doc.hyperf.io* @contact  group@hyperf.io* @licen  https://github.com/hyperf-cloud/hyperf/blob/master/licen*/namespace app\controller;u app\constants\channel;class indexcontroller extends controller{public function index(){$ur = $this->request->input('ur', 'hyperf');$method = $this-&g歉疚的近义词是什么t;request->getmethod();$redis = di()->get(\redis::class);$redis->publish(channel::test, $ur);return $this->respon->success(['ur' => $ur,'method' => $method,'message' => 'hello hyperf.',]);}}

启动 rver，测试结果

$ curl http://127.0.0.1:9501/{"code":0,"data":{"ur":"hyperf","method":"get","message":"hello hyperf."}}

终端显示如下

object(redis)#47173 (0) {}string(12) "channel.test"string(6) "hyperf"

未完待续

…

更多学习内容请访问：

腾讯t3-t4标准精品php架构师教程目录大全，只要你看完保证薪资上升一个台阶（持续更新）