Laravel Event(事件)注册和执行流程源码分析

cccdz 的个人博客 / 401 / 1 / 创建于 8个月前 / 更新于 8个月前

基于laravel10分析
我们能够知道事件会在 App\Providers\EventServiceProvider 这个服务里面注册
我们看一下这个服务里面做了什么处理
我们可以看到App\Providers\EventServiceProvider 父类中有一个register方法
我们看一下register方法中做了什么事情

public function register()
    {
        //加载boot前执行
        $this->booting(function () {
            //获取所有事件
            $events = $this->getEvents();
            //循环所有事件
            foreach ($events as $event => $listeners) {
                //这里需要去重的原因是自动发现和手动注册有相同的
                foreach (array_unique($listeners, SORT_REGULAR) as $listener) {
                    Event::listen($event, $listener);
                }
            }
            //循环订阅事件
            foreach ($this->subscribe as $subscriber) {
                Event::subscribe($subscriber);
            }
            //模型事件这里先不分析
            foreach ($this->observers as $model => $observers) {
                $model::observe($observers);
            }
        });
    }

我们先来看怎么获取所有事件的, 看一下getEvents方法做了什么处理

public function getEvents()
    {
        //这里是先判断是否有事件缓存
        if ($this->app->eventsAreCached()) {
            //这里就不分析是怎么缓存的了，可以去看一下事件缓存命令（event:cache）
            $cache = require $this->app->getCachedEventsPath();
            return $cache[get_class($this)] ?? [];
        } else {
            //主要是看这里 递归合并 和一个key放到一个数组里面
            return array_merge_recursive(
                //自动发现，找对应目录的文件，看是否有handle/__invoke方法，有就注册
                $this->discoveredEvents(),
                //这里就是拿$listen属性
                $this->listens()
            );
        }
    }

getEvents方法就分析完了
我们再来分析一下Event::listen()方法做了什么处理
首先Event这个类是静态代理类，代理的组件名称是events别名，我们能够找到这个绑定到容器的地方

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

静态代理调用listen方法就是运行的Dispatcher这个类的方法
我们看一下listen方法做了什么处理

public function listen($events, $listener = null)
 {
         //这里是判断事件是否是闭包
        if ($events instanceof Closure) {
            //如果是闭包 就反射闭包拿到第一个参数 拿到事件类名 可以有多个 比如function (Test1|Test2|Test3 $event) {}
            return collect($this->firstClosureParameterTypes($events))
                ->each(function ($event) use ($events) {
                    $this->listen($event, $events);
                });
        } 
        //判断事件是否是一个队列闭包 就是把闭包传入到queueable()辅助函数里面
        elseif ($events instanceof QueuedClosure) {
            return collect($this->firstClosureParameterTypes($events->closure))
                ->each(function ($event) use ($events) {
                    //$events->resolve()这里是和队列相关的，这里就先不分析这个了
                    $this->listen($event, $events->resolve());
                });
        } 
        //判断监听是否是一个队列闭包
        elseif ($listener instanceof QueuedClosure) {
            //这里和上面是一样的 有兴趣的小伙伴可以去看一下这个源码
            $listener = $listener->resolve();
        }

        foreach ((array) $events as $event) {
            if (str_contains($event, '*')) {
                //这里如果有带*号 则放到通配符监听数组中
                $this->setupWildcardListen($event, $listener);
            } else {
                //否则就放到listeners数组中
                $this->listeners[$event][] = $listener;
            }
        }
}

这个是resolve方法做的处理就是把闭包队列包装了一层，让队列能统一处理

public function resolve()
{
        return function (...$arguments) {
            dispatch(new CallQueuedListener(InvokeQueuedClosure::class, 'handle', [
                'closure' => new SerializableClosure($this->closure),
                'arguments' => $arguments,
                'catch' => collect($this->catchCallbacks)->map(function ($callback) {
                    return new SerializableClosure($callback);
                })->all(),
            ]))->onConnection($this->connection)->onQueue($this->queue)->delay($this->delay);
        };
}

我们看一下setupWildcardListen方法做了处理

protected function setupWildcardListen($event, $listener)
 {
         //设置到通配符数组中
        $this->wildcards[$event][] = $listener;
        //清空缓存 后面可以看到这个的作用
        $this->wildcardsCache = [];
 }

listen方法分析完了
我们再来看一下Event::subscribe()订阅方法做了什么处理

public function subscribe($subscriber)
{    
        //解析订阅 拿到订阅实例
        $subscriber = $this->resolveSubscriber($subscriber);
        //这里就是绑定的地方  拿到绑定关系
        $events = $subscriber->subscribe($this);
        //如果不是数组 代表就是用的 Event::listen去绑定关系的
        //如果是数组
        if (is_array($events)) {
            foreach ($events as $event => $listeners) {
                foreach (Arr::wrap($listeners) as $listener) {
                    //这里是一个 事件绑定方法 TestEvent::class => 'login' 这种
                    if (is_string($listener) && method_exists($subscriber, $listener)) {
                        $this->listen($event, [get_class($subscriber), $listener]);

                        continue;
                    }
                    //这里有几种形式 
                    //TestEvent::class => 'xxx@login' 
                    //TestEvent::class => [[xxx::class, 'login'],]
                    $this->listen($event, $listener);
                }
            }
        }
}

subscribe方法里面最终就是调用listen方法

有两个需要注意的地方
1.通配符事件，就是没有事件类，只有一个带有*号的名称，在listener类的方法参数就会有两个参数一个是事件名称，一个是传递的参数数组
2.不是通配符事件，但是也没有事件类的，listener类中就只能写一个参数就是传递的参数数组(正常情况这个参数是事件类对象)

事件和订阅注册就分析完了
我们再来分析一下事件执行
事件执行有两种方式event(new xxx())和xxx::dispatch()，这两种效果是一样的
最终这都是执行Dispatcher类的dispatch方法
我们看一下dispatch这个方法做了什么处理

public function dispatch($event, $payload = [], $halt = false)
{
        //这里就是拿事件名称和有效载荷(就是listener类的方法传入的参数)
        [$event, $payload] = $this->parseEventAndPayload(
            $event, $payload
        );
        ...
}

我们看一下parseEventAndPayload方法做了什么处理

protected function parseEventAndPayload($event, $payload)
{
        //这里判断是否是对象
        if (is_object($event)) {
            //拿到类名 和 有效载荷就是事件对象 不管传入的参数
            [$payload, $event] = [[$event], get_class($event)];
        }
        //如果不是对象 就直接拿事件名称 和 传入的参数当成有效载荷

        return [$event, Arr::wrap($payload)];
 }

我们看一下dispatch方法后续做了什么处理

public function dispatch($event, $payload = [], $halt = false)
    {
           ...
        //这里是判断是否应该广播，这里就不分析这个里面是怎么处理广播的了
        if ($this->shouldBroadcast($payload)) {
            $this->broadcastEvent($payload[0]);
        }

        $responses = [];
        //主要是getListeners方法
        foreach ($this->getListeners($event) as $listener) {
            $response = $listener($event, $payload);
            //这里如果传入的$halt为true  并且结果不为null 就返回当前结果 用于事件只绑定了一个监听 且想要得到监听结果的时候
            if ($halt && ! is_null($response)) {
                return $response;
            }
            //结果返回false 会终止后续的事件
            if ($response === false) {
                break;
            }
            //记录结果
            $responses[] = $response;
        }
        //返回结果 这里事件就执行完了
        return $halt ? null : $responses;
}

我们看一下getListeners方法做了什么处理

public function getListeners($eventName)
 {
         //这里做了一次合并把listeners和通配符做了一次合并
        $listeners = array_merge(
            //这里是拿eventName对应的listeners 
            $this->prepareListeners($eventName),
            //这里是如果存在缓存就先拿缓存 
            $this->wildcardsCache[$eventName] ?? $this->getWildcardListeners($eventName)
        );

        return class_exists($eventName, false)
                    //这里是判断这个类的接口类是有有对应的listeners
                    ? $this->addInterfaceListeners($eventName, $listeners)
                    : $listeners;
 }

有一个疑问，就是listeners为什么不和通配符数组一样做一个缓存，我的想到的是正常的事件一般在这个请求中只会执行一次，但是通配符可能会执行多次，所以通配符数组做了一次缓存，但是如果是在octane模式下，我觉得给listeners也加一个缓存好一点
prepareListeners和getWildcardListeners这两个方法里面都是调用了makeListener方法
我们看一下makeListener做了什么事情

 public function makeListener($listener, $wildcard = false)
 {
        //这里是xxx::class@ccc  /xxx.ccc 这种情况 
        if (is_string($listener)) {
            return $this->createClassListener($listener, $wildcard);
        }
        //这里是[xxx::class, 'ccc']/[xxx::class]的情况
        if (is_array($listener) && isset($listener[0]) && is_string($listener[0])) {
            return $this->createClassListener($listener, $wildcard);
        }
        //这里就是[new xxx(), 'xxx'] / 闭包 的情况
        return function ($event, $payload) use ($listener, $wildcard) {
            if ($wildcard) {
                return $listener($event, $payload);
            }

            return $listener(...array_values($payload));
        };
 }

我们看一下createClassListener方法做了什么处理

public function createClassListener($listener, $wildcard = false)
{
        return function ($event, $payload) use ($listener, $wildcard) {
            //判断是否是通配符情况
            if ($wildcard) {
                //把事件名称和有效载荷传入到方法里面去
                return call_user_func($this->createClassCallable($listener), $event, $payload);
            }

            $callable = $this->createClassCallable($listener);
            //这里只会把有效载荷传入进去(这里就是事件对象) 如果没有用事件类 这里就是传入参数
            return $callable(...array_values($payload));
        };
 }

看一下createClassCallable做了什么处理

protected function createClassCallable($listener)
{    
        //拿到类和方法
        [$class, $method] = is_array($listener)
                            ? $listener
                            //这里是通过@去切割
                            : $this->parseClassCallable($listener);
        //如果不存在方法就使用__invoke
        if (! method_exists($class, $method)) {
            $method = '__invoke';
        }
        //是否应该放入队列
        if ($this->handlerShouldBeQueued($class)) {
            //创建队列闭包
            return $this->createQueuedHandlerCallable($class, $method);
        }

        $listener = $this->container->make($class);
        //这里就是判断是否事务提交后执行
        return $this->handlerShouldBeDispatchedAfterDatabaseTransactions($listener)
                    //加入到事务提交后回调闭包数组里面
                    ? $this->createCallbackForListenerRunningAfterCommits($listener, $method)
                    : [$listener, $method];
 }

看一下handlerShouldBeQueued方法做了什么处理
这里面就是反射判断是否有ShouldQueue接口类

protected function handlerShouldBeQueued($class)
  {
        try {
            return (new ReflectionClass($class))->implementsInterface(
                ShouldQueue::class
            );
        } catch (Exception) {
            return false;
        }
}

看一下createQueuedHandlerCallable方法做了什么处理

protected function createQueuedHandlerCallable($class, $method)
{
        return function () use ($class, $method) {
            //这里是拿闭包传入的参数 如果是对象则clone一下 防止对象引用污染数据
            $arguments = array_map(function ($a) {
                return is_object($a) ? clone $a : $a;
            }, func_get_args());
            //这里就是判断是否想要放入队列
            if ($this->handlerWantsToBeQueued($class, $arguments)) {
                $this->queueHandler($class, $method, $arguments);
            }
            //如果不是想要加入队列 就不执行监听的业务代码了
        };
}

看一下handlerWantsToBeQueued这个方法做了什么处理

protected function handlerWantsToBeQueued($class, $arguments)
{
        //这里通过容器实例化监听者
        $instance = $this->container->make($class);
        //判断是否存在这个方法
        if (method_exists($instance, 'shouldQueue')) {
            //调用方法把参数传入进去 要么是事件类对象 要么是调用事件传入的参数(不是使用事件类的时候)
            return $instance->shouldQueue($arguments[0]);
        }

        return true;
 }

看一下queueHandler方法做了什么处理
这个方法就是拿到队列所需参数丢入到队列里面去

protected function queueHandler($class, $method, $arguments)
{
        [$listener, $job] = $this->createListenerAndJob($class, $method, $arguments);

        $connection = $this->resolveQueue()->connection(method_exists($listener, 'viaConnection')
                    ? (isset($arguments[0]) ? $listener->viaConnection($arguments[0]) : $listener->viaConnection())
                    : $listener->connection ?? null);

        $queue = method_exists($listener, 'viaQueue')
                    ? (isset($arguments[0]) ? $listener->viaQueue($arguments[0]) : $listener->viaQueue())
                    : $listener->queue ?? null;

        isset($listener->delay)
                    ? $connection->laterOn($queue, $listener->delay, $job)
                    : $connection->pushOn($queue, $job);
}

protected function createListenerAndJob($class, $method, $arguments)
{
    $listener = (new ReflectionClass($class))->newInstanceWithoutConstructor();

        return [$listener, $this->propagateListenerOptions(
        $listener, new CallQueuedListener($class, $method, $arguments)
    )];
}
//这里可以留意一下 以后事件使用异步的时候 可以看一下能够使用哪些队列参数
protected function propagateListenerOptions($listener, $job)
{
    return tap($job, function ($job) use ($listener) {
            $data = array_values($job->data);

            $job->afterCommit = property_exists($listener, 'afterCommit') ? $listener->afterCommit : null;
            $job->backoff = method_exists($listener, 'backoff') ? $listener->backoff(...$data) : ($listener->backoff ?? null);
            $job->maxExceptions = $listener->maxExceptions ?? null;
            $job->retryUntil = method_exists($listener, 'retryUntil') ? $listener->retryUntil(...$data) : null;
            $job->shouldBeEncrypted = $listener instanceof ShouldBeEncrypted;
            $job->timeout = $listener->timeout ?? null;
            $job->tries = $listener->tries ?? null;

            $job->through(array_merge(
            method_exists($listener, 'middleware') ? $listener->middleware(...$data) : [],
            $listener->middleware ?? []
        ));
    });
}

看一下handlerShouldBeDispatchedAfterDatabaseTransactions方法做了什么处理

protected function handlerShouldBeDispatchedAfterDatabaseTransactions($listener)
{
        //这里其实很简单 就是判断是否有afterCommit 且 容器里面是否绑定了db.transactions
        return ($listener->afterCommit ?? null) && $this->container->bound('db.transactions');
 }

看一下createCallbackForListenerRunningAfterCommits方法做了什么处理

 protected function createCallbackForListenerRunningAfterCommits($listener, $method)
{
        return function () use ($method, $listener) {
            $payload = func_get_args();
            //其实这里很简单 就是把这个闭包放到事务提交完回调数组里面
            $this->container->make('db.transactions')->addCallback(
                function () use ($listener, $method, $payload) {
                    $listener->$method(...$payload);
                }
            );
        };
}