laravel queue rabbitmq

ไดรเวอร์คิว RabbitMQ สำหรับ Laravel

นโยบายการสนับสนุน

เฉพาะเวอร์ชันล่าสุดเท่านั้นที่จะได้รับคุณสมบัติใหม่ การแก้ไขข้อบกพร่องจะมีให้โดยใช้รูปแบบต่อไปนี้:

เวอร์ชันแพ็คเกจ	เวอร์ชันลาร์ราเวล	แก้ไขข้อบกพร่องจนถึง
13	9	8 สิงหาคม 2023	เอกสารประกอบ

การติดตั้ง

คุณสามารถติดตั้งแพ็คเกจนี้ผ่านผู้แต่งโดยใช้คำสั่งนี้:

 composer require vladimir-yuldashev/laravel-queue-rabbitmq

แพ็คเกจจะลงทะเบียนเองโดยอัตโนมัติ

การกำหนดค่า

เพิ่มการเชื่อมต่อไปที่ config/queue.php :

นี่คือการกำหนดค่าขั้นต่ำสำหรับการเชื่อมต่อ/ไดรเวอร์ rabbitMQ ในการทำงาน

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
    
       ' driver ' => ' rabbitmq ' ,
       ' hosts ' => [
           [
               ' host ' => env ( ' RABBITMQ_HOST ' , ' 127.0.0.1 ' ),
               ' port ' => env ( ' RABBITMQ_PORT ' , 5672 ),
               ' user ' => env ( ' RABBITMQ_USER ' , ' guest ' ),
               ' password ' => env ( ' RABBITMQ_PASSWORD ' , ' guest ' ),
               ' vhost ' => env ( ' RABBITMQ_VHOST ' , ' / ' ),
           ],
           // ...
       ],

       // ...
    ],

    // ...    
],

การกำหนดค่าคิวเพิ่มเติม

เลือกที่จะเพิ่มตัวเลือกคิวให้กับการกำหนดค่าการเชื่อมต่อ ทุกคิวที่สร้างขึ้นสำหรับการเชื่อมต่อนี้ จะได้รับคุณสมบัติ

เมื่อคุณต้องการจัดลำดับความสำคัญของข้อความเมื่อมีความล่าช้า คุณสามารถทำได้โดยการเพิ่มตัวเลือกพิเศษ

• เมื่อละเว้นลำดับความสำคัญสูงสุด ลำดับความสำคัญสูงสุดจะถูกตั้งค่าเป็น 2 เมื่อใช้

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' options ' => [
            ' queue ' => [
                // ...

                ' prioritize_delayed ' =>  false ,
                ' queue_max_priority ' => 10 ,
            ],
        ],
    ],

    // ...    
],

เมื่อคุณต้องการเผยแพร่ข้อความต่อการแลกเปลี่ยนด้วยคีย์การกำหนดเส้นทาง ก็สามารถทำได้โดยการเพิ่มตัวเลือกพิเศษ

• เมื่อละเว้นการแลกเปลี่ยน RabbitMQ จะใช้การแลกเปลี่ยน amq.direct สำหรับคีย์การกำหนดเส้นทาง
• เมื่อละเว้นคีย์การกำหนดเส้นทาง คีย์การกำหนดเส้นทางตามค่าเริ่มต้นจะเป็นชื่อ queue
• เมื่อใช้ %s ในคีย์การกำหนดเส้นทาง Queue_name จะถูกแทนที่

หมายเหตุ: เมื่อใช้การแลกเปลี่ยนกับคีย์การกำหนดเส้นทาง คุณอาจสร้างคิวด้วยการผูกด้วยตนเอง

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' options ' => [
            ' queue ' => [
                // ...

                ' exchange ' => ' application-x ' ,
                ' exchange_type ' => ' topic ' ,
                ' exchange_routing_key ' => '' ,
            ],
        ],
    ],

    // ...    
],

ใน Laravel งานที่ล้มเหลวจะถูกเก็บไว้ในฐานข้อมูล แต่บางทีคุณอาจต้องการสั่งกระบวนการอื่นให้ทำบางอย่างกับข้อความด้วย เมื่อคุณต้องการสั่งให้ RabbitMQ เปลี่ยนเส้นทางข้อความที่ล้มเหลวไปยัง Exchange หรือคิวเฉพาะ ก็สามารถทำได้โดยการเพิ่มตัวเลือกพิเศษ

• เมื่อละเว้นการแลกเปลี่ยน RabbitMQ จะใช้การแลกเปลี่ยน amq.direct สำหรับคีย์การกำหนดเส้นทาง
• เมื่อละเว้น routing-key routing-key โดยค่าเริ่มต้นชื่อ queue จะถูกแทนที่ด้วย '.failed'
• เมื่อใช้ %s ในคีย์การกำหนดเส้นทาง Queue_name จะถูกแทนที่

หมายเหตุ: เมื่อใช้ failed_job exchange กับ routing-key คุณอาจต้องสร้างการแลกเปลี่ยน/คิวด้วยการผูกด้วยตนเอง

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' options ' => [
            ' queue ' => [
                // ...

                ' reroute_failed ' => true ,
                ' failed_exchange ' => ' failed-exchange ' ,
                ' failed_routing_key ' => ' application-x.%s ' ,
            ],
        ],
    ],

    // ...    
],

การสนับสนุนขอบฟ้า

เริ่มต้นด้วย 8.0 แพ็คเกจนี้รองรับ Laravel Horizon ตั้งแต่แกะกล่อง ขั้นแรก ติดตั้ง Horizon จากนั้นตั้งค่า RABBITMQ_WORKER เป็น horizon

ฮอไรซอนขึ้นอยู่กับเหตุการณ์ที่ผู้ปฏิบัติงานส่งมา เหตุการณ์เหล่านี้แจ้งให้ Horizon ทราบถึงสิ่งที่ทำกับข้อความ/งาน

Library นี้รองรับ Horizon แต่ในการกำหนดค่าคุณต้องแจ้งให้ Laravel ใช้ QueueApi ที่เข้ากันได้กับ Horizon

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        /* Set to "horizon" if you wish to use Laravel Horizon. */
       ' worker ' => env ( ' RABBITMQ_WORKER ' , ' default ' ),
    ],

    // ...    
],

ใช้คลาส RabbitMQJob ของคุณเอง

บางครั้งคุณต้องทำงานกับข้อความที่เผยแพร่โดยแอปพลิเคชันอื่น
ข้อความเหล่านั้นอาจไม่เคารพสคีมาเพย์โหลดงานของ Laravel ปัญหาเกี่ยวกับข้อความเหล่านี้คือพนักงาน Laravel จะไม่สามารถระบุงานหรือคลาสจริงที่จะดำเนินการได้

คุณสามารถขยายคลาส RabbitMQJob::class วด์อินได้ และภายในการกำหนดค่าการเชื่อมต่อคิว คุณสามารถกำหนดคลาสของคุณเองได้ เมื่อคุณระบุรหัส job ในการกำหนดค่าด้วยชื่อคลาสของคุณเอง ทุกข้อความที่ดึงมาจากนายหน้าจะถูกรวมเข้ากับคลาสของคุณเอง

ตัวอย่างสำหรับการกำหนดค่า:

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' options ' => [
            ' queue ' => [
                // ...

                ' job ' =>  App  Queue  Jobs RabbitMQJob::class,
            ],
        ],
    ],

    // ...    
],

ตัวอย่างระดับงานของคุณเอง:

 <?php

namespace App  Queue  Jobs ;

use VladimirYuldashev  LaravelQueueRabbitMQ  Queue  Jobs  RabbitMQJob as BaseJob ;

class RabbitMQJob extends BaseJob
{

    /**
     * Fire the job.
     *
     * @return void
     */
    public function fire ()
    {
        $ payload = $ this -> payload ();

        $ class = WhatheverClassNameToExecute::class;
        $ method = ' handle ' ;

        ( $ this -> instance = $ this -> resolve ( $ class ))->{ $ method }( $ this , $ payload );

        $ this -> delete ();
    }
}

หรือบางทีคุณอาจต้องการเพิ่มคุณสมบัติพิเศษให้กับเพย์โหลด:

 <?php

namespace App  Queue  Jobs ;

use VladimirYuldashev  LaravelQueueRabbitMQ  Queue  Jobs  RabbitMQJob as BaseJob ;

class RabbitMQJob extends BaseJob
{
   /**
     * Get the decoded body of the job.
     *
     * @return array
     */
    public function payload ()
    {
        return [
            ' job '  => ' WhatheverFullyQualifiedClassNameToExecute@handle ' ,
            ' data ' => json_decode ( $ this -> getRawBody (), true )
        ];
    }
}

หากคุณต้องการจัดการข้อความดิบ ไม่ใช่ในรูปแบบ JSON หรือไม่มีคีย์ 'งาน' ใน JSON คุณควรเพิ่ม stub สำหรับเมธอด getName :

 <?php

namespace App  Queue  Jobs ;

use Illuminate  Support  Facades  Log ;
use VladimirYuldashev  LaravelQueueRabbitMQ  Queue  Jobs  RabbitMQJob as BaseJob ;

class RabbitMQJob extends BaseJob
{
    public function fire ()
    {
        $ anyMessage = $ this -> getRawBody ();
        Log:: info ( $ anyMessage );

        $ this -> delete ();
    }

    public function getName ()
    {
        return '' ;
    }
}

ใช้การเชื่อมต่อของคุณเอง

คุณสามารถขยาย PhpAmqpLibConnectionAMQPStreamConnection::class หรือ PhpAmqpLibConnectionAMQPSLLConnection::class ในตัวได้ และภายในการกำหนดค่าการเชื่อมต่อ คุณสามารถกำหนดคลาสของคุณเองได้ เมื่อคุณระบุคีย์ connection ในการกำหนดค่าด้วยชื่อคลาสของคุณเอง การเชื่อมต่อทุกครั้งจะใช้คลาสของคุณเอง

ตัวอย่างสำหรับการกำหนดค่า:

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' connection ' = >  App  Queue  Connection MyRabbitMQConnection::class,
    ],

    // ...    
],

ใช้คลาส Worker ของคุณเอง

หากคุณต้องการใช้ RabbitMQQueue::class ของคุณเอง สิ่งนี้เป็นไปได้โดยขยาย VladimirYuldashevLaravelQueueRabbitMQQueueRabbitMQQueue และแจ้งให้ laravel ใช้คลาสของคุณโดยตั้งค่า RABBITMQ_WORKER เป็น AppQueueRabbitMQQueue::class

หมายเหตุ: คลาสผู้ปฏิบัติงาน ต้อง ขยาย VladimirYuldashevLaravelQueueRabbitMQQueueRabbitMQQueue

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        /* Set to a class if you wish to use your own. */
       ' worker ' =>  App  Queue RabbitMQQueue::class,
    ],

    // ...    
],

 <?php

namespace App  Queue ;

use VladimirYuldashev  LaravelQueueRabbitMQ  Queue  RabbitMQQueue as BaseRabbitMQQueue ;

class RabbitMQQueue extends BaseRabbitMQQueue
{
    // ...
}

ตัวอย่าง: การใช้งานการเชื่อมต่อใหม่

หากคุณต้องการเชื่อมต่อกับ RabbitMQ อีกครั้ง หากการเชื่อมต่อขาดหาย คุณสามารถแทนที่การเผยแพร่และวิธีการ createChannel

หมายเหตุ: นี่ไม่ใช่แนวทางปฏิบัติที่ดีที่สุด แต่เป็นเพียงตัวอย่าง

 <?php

namespace App  Queue ;

use PhpAmqpLib  Exception  AMQPChannelClosedException ;
use PhpAmqpLib  Exception  AMQPConnectionClosedException ;
use VladimirYuldashev  LaravelQueueRabbitMQ  Queue  RabbitMQQueue as BaseRabbitMQQueue ;

class RabbitMQQueue extends BaseRabbitMQQueue
{

    protected function publishBasic ( $ msg , $ exchange = '' , $ destination = '' , $ mandatory = false , $ immediate = false , $ ticket = null ): void
    {
        try {
            parent :: publishBasic ( $ msg , $ exchange , $ destination , $ mandatory , $ immediate , $ ticket );
        } catch ( AMQPConnectionClosedException | AMQPChannelClosedException ) {
            $ this -> reconnect ();
            parent :: publishBasic ( $ msg , $ exchange , $ destination , $ mandatory , $ immediate , $ ticket );
        }
    }

    protected function publishBatch ( $ jobs , $ data = '' , $ queue = null ): void
    {
        try {
            parent :: publishBatch ( $ jobs , $ data , $ queue );
        } catch ( AMQPConnectionClosedException | AMQPChannelClosedException ) {
            $ this -> reconnect ();
            parent :: publishBatch ( $ jobs , $ data , $ queue );
        }
    }

    protected function createChannel (): AMQPChannel
    {
        try {
            return parent :: createChannel ();
        } catch ( AMQPConnectionClosedException ) {
            $ this -> reconnect ();
            return parent :: createChannel ();
        }
    }
}

คิวเริ่มต้น

การเชื่อมต่อใช้คิวเริ่มต้นที่มีค่า 'ค่าเริ่มต้น' เมื่อ laravel ไม่ได้จัดเตรียมคิวไว้ เป็นไปได้ที่จะเปลี่ยนคิวเริ่มต้นโดยการเพิ่มพารามิเตอร์พิเศษในการกำหนดค่าการเชื่อมต่อ

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...
            
        ' queue ' => env ( ' RABBITMQ_QUEUE ' , ' default ' ),
    ],

    // ...    
],

การเต้นของหัวใจ

ตามค่าเริ่มต้น การเชื่อมต่อของคุณจะถูกสร้างขึ้นด้วยการตั้งค่าฮาร์ตบีทเป็น 0 คุณสามารถเปลี่ยนการตั้งค่าฮาร์ตบีทได้โดยเปลี่ยนการกำหนดค่า

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' options ' => [
            // ...

            ' heartbeat ' => 10 ,
        ],
    ],

    // ...    
],

SSL ที่ปลอดภัย

หากคุณต้องการการเชื่อมต่อที่ปลอดภัยกับเซิร์ฟเวอร์ rabbitMQ คุณจะต้องเพิ่มตัวเลือกการกำหนดค่าพิเศษเหล่านี้

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' secure ' = > true ,
        ' options ' => [
            // ...

            ' ssl_options ' => [
                ' cafile ' => env ( ' RABBITMQ_SSL_CAFILE ' , null ),
                ' local_cert ' => env ( ' RABBITMQ_SSL_LOCALCERT ' , null ),
                ' local_key ' => env ( ' RABBITMQ_SSL_LOCALKEY ' , null ),
                ' verify_peer ' => env ( ' RABBITMQ_SSL_VERIFY_PEER ' , true ),
                ' passphrase ' => env ( ' RABBITMQ_SSL_PASSPHRASE ' , null ),
            ],
        ],
    ],

    // ...    
],

เหตุการณ์หลังจากฐานข้อมูลคอมมิต

เพื่อสั่งให้พนักงาน Laravel จัดส่งกิจกรรมหลังจากที่ฐานข้อมูลทั้งหมดเสร็จสิ้น

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' after_commit ' => true ,
    ],

    // ...    
],

การเชื่อมต่อที่ขี้เกียจ

ตามค่าเริ่มต้น การเชื่อมต่อของคุณจะถูกสร้างขึ้นเป็นการเชื่อมต่อแบบ Lazy หากคุณไม่ต้องการให้การเชื่อมต่อล่าช้าด้วยเหตุผลบางประการ คุณสามารถปิดการเชื่อมต่อได้โดยตั้งค่าการกำหนดค่าต่อไปนี้

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' lazy ' = > false ,
    ],

    // ...    
],

โปรโตคอลเครือข่าย

ตามค่าเริ่มต้น โปรโตคอลเครือข่ายที่ใช้สำหรับการเชื่อมต่อคือ tcp หากคุณต้องการใช้โปรโตคอลเครือข่ายอื่นด้วยเหตุผลบางประการ คุณสามารถเพิ่มมูลค่าพิเศษในตัวเลือกการกำหนดค่าของคุณได้ โปรโตคอลที่ใช้ได้: tcp , ssl , tls

 ' connections ' => [
    // ...

    ' rabbitmq ' => [
        // ...

        ' network_protocol ' => ' tcp ' ,
    ],

    // ...    
],

การสนับสนุนออกเทน

ตั้งแต่ 13.3.0 เป็นต้นไป แพ็คเกจนี้รองรับ Laravel Octane ตั้งแต่แกะกล่อง ขั้นแรก ติดตั้ง Octane และอย่าลืมอุ่นการเชื่อมต่อ 'rabbitmq' ในการกำหนดค่าออกเทน

ดู: #460 (ความเห็น)

การใช้งาน Laravel

เมื่อคุณกำหนดค่าเสร็จแล้ว คุณจะสามารถใช้ Laravel Queue API ได้ หากคุณใช้ไดรเวอร์คิวอื่น คุณไม่จำเป็นต้องเปลี่ยนแปลงอะไรอีก หากคุณไม่ทราบวิธีใช้ Queue API โปรดดูเอกสารประกอบอย่างเป็นทางการของ Laravel: http://laravel.com/docs/queues

การใช้งานลูเมน

สำหรับการใช้งาน Lumen ผู้ให้บริการควรลงทะเบียนด้วยตนเองดังนี้ใน bootstrap/app.php :

 $ app -> register ( VladimirYuldashev  LaravelQueueRabbitMQ LaravelQueueRabbitMQServiceProvider::class);

การบริโภคข้อความ

มีสองวิธีในการบริโภคข้อความ

1. queue:work คำสั่งซึ่งเป็นคำสั่งในตัวของ Laravel คำสั่งนี้ใช้ basic_get ใช้สิ่งนี้หากคุณต้องการใช้หลายคิว

2. คำสั่ง rabbitmq:consume ซึ่งจัดทำโดยแพ็คเกจนี้ คำสั่งนี้ใช้ basic_consume และมีประสิทธิภาพมากกว่า basic_get ประมาณ ~2x แต่ไม่รองรับหลายคิว

การทดสอบ

ตั้งค่า RabbitMQ โดยใช้ docker-compose :

docker compose up -d

หากต้องการรันชุดทดสอบ คุณสามารถใช้คำสั่งต่อไปนี้:

 # To run both style and unit tests.
composer test

# To run only style tests.
composer test:style

# To run only unit tests.
composer test:unit

หากคุณได้รับข้อผิดพลาดจากการทดสอบสไตล์ คุณสามารถแก้ไขปัญหาส่วนใหญ่ได้โดยอัตโนมัติ หากไม่ใช่ทั้งหมดด้วยคำสั่งต่อไปนี้:

composer fix:style

ผลงาน

คุณสามารถมีส่วนร่วมในแพ็คเกจนี้ได้โดยการค้นหาจุดบกพร่องและปัญหาในการเปิด โปรดเพิ่มเวอร์ชันของแพ็คเกจที่คุณสร้างคำขอดึงหรือปัญหา (เช่น [5.2] ข้อผิดพลาดร้ายแรงในงานล่าช้า)