laravel队列任务简单使用

追影

1. <?php
   
2. 
   
3. namespace App\Jobs;
   
4. 
   
5. use App\Models\AiMailTask;
   
6. use App\Models\AiMailTaskDetail;
   
7. use App\Models\Email\EMailBind;
   
8. use App\Models\Email\EMailSubscribe;
   
9. use App\Repositories\AiMailTaskDetailRepository;
   
10. use App\Repositories\AiPotentialCustomerAcquisitionTaskDetailRepository;
    
11. use Carbon\Carbon;
    
12. use Exception;
    
13. use Illuminate\Bus\Queueable;
    
14. use Illuminate\Contracts\Queue\ShouldQueue;
    
15. use Illuminate\Foundation\Bus\Dispatchable;
    
16. use Illuminate\Queue\InteractsWithQueue;
    
17. use Illuminate\Queue\SerializesModels;
    
18. use Cache;
    
19. use Log;
    
20. 
    
21. /**
    
22. * 创建AI邮件群发任务
    
23. */
    
24. class AiEmailWillTaskJob implements ShouldQueue
    
25. {
    
26.     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    
27. 
    
28.     protected $param;
    
29. 
    
30.     /**
    
31.      * Create a new job instance.
    
32.      *
    
33.      * @return void
    
34.      */
    
35.     public function __construct($params)
    
36.     {
    
37.         $this->connection = 'ai_mail_sub_task_create';
    
38.         $this->param = $params;
    
39.     }
    
40. 
    
41.     /**
    
42.      * Execute the job.
    
43.      *
    
44.      * @return true|void
    
45.      */
    
46.     public function handle()
    
47.     {
    
48.         $param = $this->param;
    
49.         try {
    
50.             $post_id     = $param['post_id'];
    
51.             $user_id     = $param['user_id'];
    
52.             $task_id     = $param['task_id'];
    
53.             $bind_name   = $param['bind_name'];
    
54.             $bind_email  = $param['bind_email'];
    
55.             $email_type  = $param['email_type'];
    
56.             $custom_list = $param['custom_list'];
    
57.             $removal_day = $param['removal_day'];
    
58.             if (!$post_id || !$user_id || !$task_id || (!$email_type && !$custom_list)) {
    
59.                 Log::error("邮件群发参数不完整");
    
60.                 return 0;
    
61.             }
    
62.             // 加个创建子任务标志
    
63.             $cache_key = 'ai_mail_send_create_subtask_' . $task_id;
    
64.             Cache::put($cache_key, '1');
    
65.             if (!is_array($email_type) && $email_type) {
    
66.                 $email_type = [$email_type];
    
67.             }
    
68.             if ($email_type) { // AI潜客采集列表
    
69.                 $customers = app(AiPotentialCustomerAcquisitionTaskDetailRepository::class)->getMultiTaskValidEmails($email_type);
    
70. 
    
71.                 $all_mailboxes          = $customers->pluck('mailbox')->toArray(); //待发送的所有目标邮箱
    
72.                 $history_send_mailboxes = [];
    
73.                 // 过滤多少天内发送过的邮箱
    
74.                 if ($removal_day) {
    
75.                     $history_send_mailboxes = $this->filterSentMailboxes($all_mailboxes, $removal_day, $user_id);
    
76.                 }
    
77.                 foreach ($customers as $customer) {
    
78.                     $email = $customer->mailbox;
    
79.                     if (empty($email)) continue;
    
80.                     // 在$removal_day天发过，不发送该邮箱
    
81.                     if (in_array($email, $history_send_mailboxes)) {
    
82.                         continue;
    
83.                     }
    
84.                     $name = $customer->name;
    
85.                     if (empty($name)) {
    
86.                         $temp = explode("@", $email);
    
87.                         $name = $temp[0];
    
88.                     }
    
89.                     $this->createSubTask($task_id, $post_id, $user_id, $bind_name, $bind_email, $email, $name, $customer->id);
    
90.                 }
    
91.             }
    
92.             if ($custom_list) { // 添加自定义邮箱（自定义列表）
    
93.                 $custom_list = str_replace(array("\r\n", "\r", "\n"), '加入异步队列
    
94. [code]                //加入计划任务队列
    
95.                 if ($data['email_type'] || $data['custom_list'] || $data['subscribe']) {
    
96.                     $param = ['email_type' => $data['email_type'], 'custom_list' => $data['custom_list'], 'task_id' => $aiMailTask->id, 'post_id' => $postId, 'user_id' => $userId, 'bind_name' => ($bindMailbox->name ?? null), 'bind_email' => ($bindMailbox->email ?? null), 'removal_day' => $aiMailTask->removal_day,'subscribe'=>$data['subscribe'],'filter_subscribe'=>$attributes['filter_subscribe']];
    
97.                     AiEmailWillTaskJob::dispatch($param);
    
98.                 }

复制代码

, trim($custom_list));
                $customList  = explode('加入异步队列

1.                 //加入计划任务队列
   
2.                 if ($data['email_type'] || $data['custom_list'] || $data['subscribe']) {
   
3.                     $param = ['email_type' =&gt; $data['email_type'], 'custom_list' =&gt; $data['custom_list'], 'task_id' =&gt; $aiMailTask-&gt;id, 'post_id' =&gt; $postId, 'user_id' =&gt; $userId, 'bind_name' =&gt; ($bindMailbox-&gt;name ?? null), 'bind_email' =&gt; ($bindMailbox-&gt;email ?? null), 'removal_day' =&gt; $aiMailTask-&gt;removal_day,'subscribe'=&gt;$data['subscribe'],'filter_subscribe'=&gt;$attributes['filter_subscribe']];
   
4.                     AiEmailWillTaskJob::dispatch($param);
   
5.                 }

复制代码

, $custom_list);
                $name_and_mailboxes = [];
                foreach ($customList as $customerId) {
                    $customerId = preg_replace('/^[\s ]+|[\s ]+$/u', '', $customerId);// 去掉全角空格
                    $customerId = str_replace(['<','>'], '', $customerId);// 去掉全角空格
                    $customerId = trim($customerId); // 去掉两边的空格
                    $temp = explode('|', $customerId);
                    $temp = array_map('trim', $temp);// 去掉两边的空格
                    $name = '';
                    $to   = '';
                    if (count($temp) == 1 && filter_var($customerId, FILTER_VALIDATE_EMAIL)) {
                        $to   = $customerId;
                        $arr  = explode("@", $to);
                        $name = $arr[0];
                    } else if (count($temp) == 2 && filter_var($temp[1], FILTER_VALIDATE_EMAIL)) {
                        $to   = $temp[1];
                        $name = $temp[0];
                    }
                    if (!$to) continue;
                    if (empty($name)) {
                        $name = explode("@", $to)[0];
                    }
                    $name_and_mailboxes[$to] = $name;
                }

                if ($name_and_mailboxes) {
                    $history_send_mailboxes = [];
                    if ($removal_day) {
                        $history_send_mailboxes = $this->filterSentMailboxes(array_keys($name_and_mailboxes), $removal_day, $user_id);
                    }

                    foreach ($name_and_mailboxes as $mailbox => $name) {
                        // 过滤多少天内发送过的邮箱
                        if (in_array($mailbox, $history_send_mailboxes)) {
                            continue;
                        }
                        $this->createSubTask($task_id, $post_id, $user_id, $bind_name, $bind_email, $mailbox, $name);
                    }
                }
            }

            if (!empty($param['subscribe'])) {// 订阅邮箱
                $bindEmails = $this->getBindEmail();
                $p = 1;
                while ($p) {
                    $subModel = EMailSubscribe::query()->whereIn('c_email', $bindEmails);//->get(['u_email']);
                    // 过滤已取消订阅的
                    if (!empty($param['filter_subscribe'])) {
                        $subModel->where('status', 's');
                    }
                    $subModel = $subModel->forPage($p, 1000)->get(['u_email'])->pluck('u_email')->toArray();

                    foreach ($subModel as $item) {
                        // 邮箱存在，如果收件人姓名(name)不存在，name就取@前面的值，不可能为空
                        $this->createSubTask($task_id, $post_id, $user_id, $bind_name, $bind_email, $item, explode('@', $item)[0]);
                    }
                    // 没有了，跳出
                    if (count($subModel) < 1000) {
                        break;
                    }
                }
            }
            // 创建完成更新主任务的发送总数
            $task = AiMailTask::find($task_id);
            $count = $task->taskDetails->count();
            $updateParams = ['send_total_num' => $count];
            if ($count == 0) {
                $updateParams['status'] = AiMailTask::FAILED_STATUS;
                $updateParams['remark'] = '过滤后未获取到收件邮箱';
            }
            $task->update($updateParams);
            Cache::forget($cache_key);
            return true;
        } catch (Exception $e) {
            Log::error("AI发邮件-写入邮件队列失败：" . $e, ['params' => $param]);
        }
    }

    /**
     * 获取发送邮箱
     * @return array
     * @author：dc
     * @time 2024/8/13 16:21
     */
    public function getBindEmail(){
        if($this->param['bind_email']){
            return [$this->param['bind_email']];
        }else{
            if(empty($this->bindEmail)){
                $this->bindEmail = EMailBind::query()->where('post_id',$this->param['post_id'])->get(['email'])->pluck('email')->toArray();
            }
            return $this->bindEmail;
        }
    }

    /**
     * @param $task_id
     * @param $post_id
     * @param $user_id
     * @param $bind_name
     * @param $bind_email
     * @param $mailbox
     * @param $name
     * @param int $recipient_id
     * @return void
     */
    public function createSubTask($task_id, $post_id, $user_id, $bind_name, $bind_email, $mailbox, $name, $recipient_id = 0): void
    {
        // 是否要过滤掉退订的邮箱
        if(empty($this->param['subscribe']) && !empty($this->param['filter_subscribe'])){
            $unsubscribe = EMailSubscribe::query()
                ->whereIn('c_email',$this->getBindEmail())
                ->where('status','d')->count();
            if($unsubscribe){
                return ;
            }
        }

        // 项目包含指定的域名的邮箱则不创建发件任务
        if ($post_id == 544429) { // 浙江大铭汽车零部件有限公司
            $filterDomains = ['Bilsteingroup.com', 'bilsteingroup.com', 'driv.com', 'autozone.com', 'motorcarparts.com',
                'atp-autoteile.de', 'valeryd.se', 'smartparts.eu', 'nexusautomotive.eu', 'sapco.com', 'ntn-snr.fr',
                'mgadistribution.fr', 'optimal-germany.com'];
            if (in_array(explode('@', $mailbox)[1], $filterDomains)) {
                return;
            }
        }

        $insert['task_id']       = $task_id;
        $insert['post_id']       = $post_id;
        $insert['user_id']       = $user_id;
        $insert['send_nickname'] = $bind_name;
        $insert['send_mailbox']  = $bind_email;
        $insert['to']            = $mailbox;
        $insert['name']          = $name;
        $insert['recipient_id']  = $recipient_id;
        $insert['status']        = AiMailTaskDetail::CREATED_STATUS;
        app(AiMailTaskDetailRepository::class)->createDetailTask($insert);
    }

    /**
     * 过滤$removal_day天内已发送过的邮箱
     * @param array $object_mailboxes 待发送的所有目标邮箱
     * @param $removal_day
     * @param $user_id
     * @return array
     */
    public function filterSentMailboxes(array $object_mailboxes, $removal_day, $user_id)
    {
        $history_send_mailboxes = [];
        // 查询多少天内发送过的邮箱
        if ($removal_day) {
            $history_send_mailboxes = AiMailTaskDetail::where(['user_id' => $user_id])
                ->where('created_at', '>', Carbon::now()->subDays($removal_day))
                ->where('status', AiMailTaskDetail::SUCCESS_STATUS)
                ->whereIn('to', $object_mailboxes)
                ->pluck('to')
                ->toArray();
            $history_send_mailboxes = array_unique($history_send_mailboxes);
        }
        return $history_send_mailboxes;
    }
}
[/code]加入异步队列
[        DISCUZ_CODE_9        ]