基于 Redis 消息队列实现 Laravel 事件监听及底层源码探究

sail artisan make:event PostViewed

<?php

namespace App\Events;

use App\Models\Post;
use Illuminate\Broadcasting\Channel;
use Illuminate\Broadcasting\InteractsWithSockets;
use Illuminate\Broadcasting\PrivateChannel;
use Illuminate\Foundation\Events\Dispatchable;
use Illuminate\Queue\SerializesModels;

class PostViewed
{
    use Dispatchable, InteractsWithSockets, SerializesModels;

    public Post $post;

    /**
     * Create a new event instance.
     *
     * @param Post $post
     */
    public function __construct(Post $post)
    {
        $this->post = $post;
    }

    /**
     * Get the channels the event should broadcast on.
     *
     * @return Channel|array
     */
    public function broadcastOn()
    {
        return new PrivateChannel('channel-name');
    }
}

php artisan make:listener IncreasePostViews

<?php

namespace App\Listeners;

use App\Events\PostViewed;
use Illuminate\Support\Facades\Redis;

class IncreasePostViews
{
    /**
     * Create the event listener.
     *
     * @return void
     */
    public function __construct()
    {
        //
    }

    /**
     * Handle the event.
     *
     * @param PostViewed $event
     * @return void
     */
    public function handle(PostViewed $event)
    {
        if ($event->post->increment('views')) {
            Redis::zincrby('popular_posts', 1, $event->post->id);
        }
    }
}

protected $listen = [
    ...
    PostViewed::class => [
        IncreasePostViews::class
    ],
];

// 浏览文章
public function show($id)
{
    $post = $this->postRepo->getById($id);
    // 触发文章浏览事件
    event(new PostViewed($post));
    return "Show Post #{$post->id}, Views: {$post->views}";
}

...
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Queue\InteractsWithQueue;

class IncreasePostViews implements ShouldQueue
{
    use InteractsWithQueue;

    ...
}

public string $queue = 'events';

public function listen($events, $listener = null)
{
    if ($events instanceof Closure) {
        return $this->listen($this->firstClosureParameterType($events), $events);
    } elseif ($events instanceof QueuedClosure) {
        return $this->listen($this->firstClosureParameterType($events->closure), $events->resolve());
    } elseif ($listener instanceof QueuedClosure) {
        $listener = $listener->resolve();
    }

    foreach ((array) $events as $event) {
        if (Str::contains($event, '*')) {
            $this->setupWildcardListen($event, $listener);
        } else {
            $this->listeners[$event][] = $this->makeListener($listener);
        }
    }
}

function event(...$args)
{
    return app('events')->dispatch(...$args);
}

public function dispatch($event, $payload = [], $halt = false)
{
    [$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 ($halt && ! is_null($response)) {
            return $response;
        }

        if ($response === false) {
            break;
        }

        $responses[] = $response;
    }

    return $halt ? null : $responses;
}

$response = $listener($event, $payload);

public function makeListener($listener, $wildcard = false)
{
    if (is_string($listener)) {
        return $this->createClassListener($listener, $wildcard);
    }

    if (is_array($listener) && isset($listener[0]) && is_string($listener[0])) {
        return $this->createClassListener($listener, $wildcard);
    }

    return function ($event, $payload) use ($listener, $wildcard) {
        if ($wildcard) {
            return $listener($event, $payload);
        }

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

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));
    };
}

protected function createClassCallable($listener)
{
    [$class, $method] = is_array($listener)
                        ? $listener
                        : $this->parseClassCallable($listener);

    if (! method_exists($class, $method)) {
        $method = '__invoke';
    }

    if ($this->handlerShouldBeQueued($class)) {
        return $this->createQueuedHandlerCallable($class, $method);
    }

    return [$this->container->make($class), $method];
}

protected function createQueuedHandlerCallable($class, $method)
{
    return function () use ($class, $method) {
        $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);
        }
    };
}

$response = $listener($event, $payload);