laravel广播源码分析

深藏阁楼爱情的钟 2023-06-22 03:22 56阅读 0赞

# 概述 #

上篇文章写的是 在laravel-vue前后分离的项目中 使用laravel-echo-serve 的服务来实现websocket 通信的,接下来就分析larave中的广播功能的源码分析

大概流程是当触发了 事件之后 发现是广播事件(实现了ShouldBroadcast这个接口) 然后 该广播事件push到队列中. 队列消费者 执行该事件 通过redis publish 对 redis 进行了频道的信息发布

# 分析源码 #

## 广播系统服务启动 ##

broadcast(new \App\Events\H2pDepositCallbackEvent($user));

通过助手函数broadcast来触发这个事件

function broadcast($event = null)
        {
            return app(BroadcastFactory::class)->event($event);
        }

解析BroadcastFactory这个类然后调用这个对象的event()方法.  
这个类 是啥时注入到容器中呢,接下来分析广播系统的注入

### 服务注册 ###

广播服务的注册 就是想ioc容器中注册广播门面类 也就是 BroadcastManager  
Illuminate\\Broadcasting\\BroadcastServiceProvider

public function register()
       {
           $this->app->singleton(BroadcastManager::class, function ($app) {
               return new BroadcastManager($app);
           });
    
           $this->app->singleton(BroadcasterContract::class, function ($app) {
               return $app->make(BroadcastManager::class)->connection();
           });
    
           $this->app->alias(
               BroadcastManager::class, BroadcastingFactory::class
           );
       }

除了注册BroadcastManager , BroadcastServiceProvider还对广播驱动进行了启动

public function connection($driver = null)
       {
           return $this->driver($driver);
       }
    public function driver($name = null)
       {
           $name = $name ?: $this->getDefaultDriver();
    
           return $this->drivers[$name] = $this->get($name);
       }
    protected function get($name)
       {
           return $this->drivers[$name] ?? $this->resolve($name);
       }
    protected function resolve($name)
       {
           $config = $this->getConfig($name);
    
           if (is_null($config)) {
               throw new InvalidArgumentException("Broadcaster [{$name}] is not defined.");
           }
    
           if (isset($this->customCreators[$config['driver']])) {
               return $this->callCustomCreator($config);
           }
    
           $driverMethod = 'create'.ucfirst($config['driver']).'Driver';
    
           if (! method_exists($this, $driverMethod)) {
               throw new InvalidArgumentException("Driver [{$config['driver']}] is not supported.");
           }
    
           return $this->{$driverMethod}($config);
       }
    这里已redis为驱动为例
    调用createRedisDriver方法 返回redis实例
    protected function createRedisDriver(array $config)
       {
           return new RedisBroadcaster(
               $this->app->make('redis'), $config['connection'] ?? null
           );
       }

分析完 BroadcastManager这个门面类 接下来 调用这个类里的event()方法

/**
        * Begin broadcasting an event.
        *
        * @param  mixed|null  $event
        * @return \Illuminate\Broadcasting\PendingBroadcast|void
        */
       public function event($event = null)
       {
           return new PendingBroadcast($this->app->make('events'), $event);
       }

$this->app->make(‘events’) 分析下这个 解析events

### 事件服务注册 ###

Illuminate\\Events\\EventServiceProvider

public function register()
        {
            $this->app->singleton('events', function ($app) {
                return (new Dispatcher($app))->setQueueResolver(function () use ($app) {
                    return $app->make(QueueFactoryContract::class);
                });
            });
        }

可以看出 Dispatcher 就是 events 真正的实现类

### 触发事件 ###

Illuminate\\Broadcasting\\PendingBroadcast  
这个类的析构方法中调用了事件对象的dispatch方法

public function __destruct()
        {
            $this->events->dispatch($this->event);
        }

事件的触发可以利用事件名，或者事件类的实例：  
Illuminate\\Events\\Dispatcher

/**
         * Fire an event and call the listeners.
         *
         * @param  string|object  $event
         * @param  mixed  $payload
         * @param  bool  $halt
         * @return array|null
         */
        public function dispatch($event, $payload = [], $halt = false)
        {
            // When the given "event" is actually an object we will assume it is an event
            // object and use the class as the event name and this event itself as the
            // payload to the handler, which makes object based events quite simple.
    
            list($event, $payload) = $this->parseEventAndPayload(
                $event, $payload
            );
    
            if ($this->shouldBroadcast($payload)) {
                $this->broadcastEvent($payload[0]);
            }
    
            $responses = [];
    
            foreach ($this->getListeners($event) as $listener) {
                $response = $listener($event, $payload);
    
                // If a response is returned from the listener and event halting is enabled
                // we will just return this response, and not call the rest of the event
                // listeners. Otherwise we will add the response on the response list.
                if ($halt && ! is_null($response)) {
                    return $response;
                }
    
                // If a boolean false is returned from a listener, we will stop propagating
                // the event to any further listeners down in the chain, else we keep on
                // looping through the listeners and firing every one in our sequence.
                if ($response === false) {
                    break;
                }
    
                $responses[] = $response;
            }
    
            return $halt ? null : $responses;
        }

parseEventAndPayload 函数利用传入参数是事件名还是事件类实例来确定监听类函数的参数：

protected function parseEventAndPayload($event, $payload)
        {
            if (is_object($event)) {
    
                list($payload, $event) = [[$event], get_class($event)];
            }
    
            return [$event, Arr::wrap($payload)];
        }

如果是事件类的实例，那么监听函数的参数就是事件类自身；如果是事件类名，那么监听函数的参数就是触发事件时传入的参数。

获得事件与参数后，就要获取监听类：

public function getListeners($eventName)
        {
            $listeners = $this->listeners[$eventName] ?? [];
    
            $listeners = array_merge(
                $listeners,
                $this->wildcardsCache[$eventName] ?? $this->getWildcardListeners($eventName)
            );
    
            return class_exists($eventName, false)
                        ? $this->addInterfaceListeners($eventName, $listeners)
                        : $listeners;
        }

寻找监听类的时候，也要从通配符监听器中寻找：

protected function getWildcardListeners($eventName)
    {
        $wildcards = [];
        foreach ($this->wildcards as $key => $listeners) {
            if (Str::is($key, $eventName)) {
                $wildcards = array_merge($wildcards, $listeners);
            }
        }
        return $wildcards;
    }

继续看 $this->shouldBroadcast

/**
    	确认是否为广播事件
         * Determine if the payload has a broadcastable event.
         *
         * @param  array  $payload
         * @return bool
         */
        protected function shouldBroadcast(array $payload)
        {
            
            return isset($payload[0]) &&
                   $payload[0] instanceof ShouldBroadcast &&
                   $this->broadcastWhen($payload[0]);
        }
        /**
         * Broadcast the given event class.
         *
         * @param  \Illuminate\Contracts\Broadcasting\ShouldBroadcast  $event
         * @return void
         */
        protected function broadcastEvent($event)
        {
            $this->container->make(BroadcastFactory::class)->queue($event);
        }

可见，关键之处在于 BroadcastManager 的 quene 方法：  
Illuminate\\Broadcasting\\BroadcastManager

/**
    	将广播事件加入到队列中
         * Queue the given event for broadcast.
         *
         * @param  mixed  $event
         * @return void
         */
        public function queue($event)
        {
            $connection = $event instanceof ShouldBroadcastNow ? 'sync' : null;
    
            if (is_null($connection) && isset($event->connection)) {
                $connection = $event->connection;
            }
    
            $queue = null;
    
            if (method_exists($event, 'broadcastQueue')) {
                $queue = $event->broadcastQueue();
            } elseif (isset($event->broadcastQueue)) {
                $queue = $event->broadcastQueue;
            } elseif (isset($event->queue)) {
                $queue = $event->queue;
            }
    
            $this->app->make('queue')->connection($connection)->pushOn(
                $queue, new BroadcastEvent(clone $event)
            );
        }

可见，quene 方法将广播事件包装为事件类，并且通过队列发布 当消息队列消费消息的时候 会执行事件类中的 handle()方法  
之前的广播事件类里是没有handel()的 所以queue方法中将广播事件包装为了事件类

### queue注册 ###

Illuminate\\Queue\\QueueServiceProvider

public function register()
        {
            $this->registerManager();
    
            $this->registerConnection();
    
            $this->registerWorker();
    
            $this->registerListener();
    
            $this->registerFailedJobServices();
        }

registerManager 负责注册队列服务的门面类：

/**
         * Register the queue manager.
         *
         * @return void
         */
        protected function registerManager()
        {
            $this->app->singleton('queue', function ($app) {
                // Once we have an instance of the queue manager, we will register the various
                // resolvers for the queue connectors. These connectors are responsible for
                // creating the classes that accept queue configs and instantiate queues.
                return tap(new QueueManager($app), function ($manager) {
                    $this->registerConnectors($manager);
                });
            });
        }

laravel tap() 助手函数 tap($value,  c a l l b a c k ) 将 callback) 将 callback)将value 作为回调函数的参数 执行回调函数 最终返回值 将 QueueManager这个类的实例 当做 回调函数的参数

**
         * Register the connectors on the queue manager.
         *
         * @param  \Illuminate\Queue\QueueManager  $manager
         * @return void
         */
        public function registerConnectors($manager)
        {
            foreach (['Null', 'Sync', 'Database', 'Redis', 'Beanstalkd', 'Sqs'] as $connector) {
                $this->{"register{$connector}Connector"}($manager);
            }
        }
        //已redis为例
         /**
         * Register the Redis queue connector.
         *
         * @param  \Illuminate\Queue\QueueManager  $manager
         * @return void
         */
        protected function registerRedisConnector($manager)
        {
            $manager->addConnector('redis', function () {
                return new RedisConnector($this->app['redis']);
            });
        }
    
     Illuminate\Queue\QueueManager
    public function addConnector($driver, Closure $resolver)
        {
            $this->connectors[$driver] = $resolver;
        }
    
        QueueManager 是队列服务的总门面，提供一切与队列相关的操作接口。QueueManager 中有一个成员变量 $connectors，该成员变量中存储着所有 laravel 支持的底层队列服务：’Database’, ‘Redis’, ‘Beanstalkd’, ‘Sqs’。
        成员变量 $connectors 会被存储各种驱动的 connector，例如 RedisConnector、SqsConnector、DatabaseConnector、BeanstalkdConnector。

registerConnection 底层队列连接服务  
接下来，就要连接实现队列的底层服务了，例如 redis  
registerWorker 消费者服务注册  
消费者的注册服务会返回 Illuminate\\Queue\\Worker 类：

protected function registerWorker()
        {
            $this->app->singleton('queue.worker', function () {
                return new Worker(
                    $this->app['queue'], $this->app['events'], $this->app[ExceptionHandler::class]
                );
            });
        }

queue服务注册完了

$this->app->make('queue')->connection($connection)->pushOn(
                $queue, new BroadcastEvent(clone $event)
            );

这里redis为例子 那么就相当于 调用 Illuminate\\Queue\\RedisQueue 里的方法 实际上是RedisQueue 继承的Illuminate\\Queue\\Queue这个抽象类里的pushOn()方法

Illuminate\Queue\RedisQueue
    public function pushOn($queue, $job, $data = '')
        {
            return $this->push($job, $data, $queue);
        }

使用redis 将job push到队列中