使用workerman/redis-queue发送邮件会出现多次发送，请问如何处理？

xtn 1天前

问题描述

再使用workerman/redis-queue循环发送邮件，比如有4个邮件，在subscribe里是能打印四个邮件，但是每个邮箱会收到四条一模一样的邮件，按道理的是话每个邮箱只应该收到一条邮件才对,请问这个要如何处理？谢谢

程序代码或配置

<?php

namespace app\command;

use app\common\service\email\EmailService;
use think\console\Command;
use think\console\Input;
use think\console\input\Argument;
use think\console\input\Option;
use think\console\Output;
use Workerman\RedisQueue\Client;
use Workerman\Worker;

class JclEmail extends Command
{
    protected function configure()
    {
        $this->setName('jcl_email')->setDescription('Run Redis Queue Worker')
            ->addOption('d', null, Option::VALUE_NONE, '守護進程')
            ->addArgument('action', Argument::OPTIONAL, 'start|stop|restart|reload|status', 'start');
    }

    protected function execute(Input $input, Output $output)
    {
        $server = new EmailService();
        $worker = new Worker();
        $worker->onWorkerStart = function () use ($output) {
            $client = new Client('redis://127.0.0.1:6379', [
                'db' => 1
            ]);
            // 订阅 sendEmail
            $client->subscribe('sendEmail', function ($data) use ($output) {
//                var_export($data);
//                $server::init()->sendEmail($data);
                try {
                    EmailService::init()->sendEmail($data);
                    $output->writeln("Email sent successfully ");
                } catch (\Throwable $e) {
                    $output->writeln("Error sending email to " . $e->getMessage());
                }
            });
            // 消费失败触发的回调(可选)
            $client->onConsumeFailure(function (\Throwable $exception, $package) {
                //\think\facade\Log::log("队列 " . $package['queue'] . " 消费失败:".$exception->getMessage());
                var_export("队列 " . $package['queue'] . " 消费失败:" . $exception->getMessage());
            });
        };

        Worker::runAll();
    }

}

重现问题的步骤

    //这里调用发送
  event(new SendEmailEvent('register', null, [
                    'name' => $this->siteInfo['name'],
                    'logo' => $this->siteInfo['logo'],
                    'account' => $params['account'],
                    'email' => [$params['email']],
                ]));

<?php
declare (strict_types=1);

namespace app\event;

use app\adminapi\logic\DemandQuoteLogic;
use app\common\model\Demand;
use app\common\model\DemandQuote;
use app\common\model\Order;
use app\common\model\user\User;
use app\common\service\ConfigService;
use app\common\service\redis\RedisService;

class SendEmailEvent
{
    public function __construct($type, $emails = [], $params = [])
    {
//        $adminEmails = ConfigService::get('email', 'admin_emails');
//        $adminEmails = explode(',', $adminEmails);
        $adminEmails = [];
        $subject = '';
        $body = '';

        $siteLogo = ConfigService::get('website', 'pc_logo');
        $siteName = ConfigService::get('website', 'pc_title');

        switch ($type) {
            //如果是注册
            case 'register':
                $subject = $params['name'] . '新用户注册邮箱提醒';
                $body = "<div>
                <div><img style='width: 50px;height: 40px;' src='" . $params['logo'] . "'/></div>
                <h4>新用户注册信息</h4>
                <p>用户名：" . $params['account'] . "</p>
                <p>邮箱：" . $params['emails'][0] . "</p>
                </div>";
                break;
            case 'companyAuthAfter':
                $subject = '新增企业认证';
                $body = "<div>
                <div><img style='width: 50px;height: 40px;' src='" . $siteLogo . "'/></div>
                <h4>企业已完成认证，请进入平台查看相关信息</h4>
                </div>";
                break;

        }
        $email_list = [];
        if (is_array($emails) && count($emails) > 0) {
            $email_list = array_merge($adminEmails, $emails);
        }

        foreach ($email_list as $email) {
            if (filter_var($email, FILTER_VALIDATE_EMAIL)) {
                (new RedisService())->redis_queue_send('sendEmail', [
                    'email' => $email,
                    'subject' => $subject,
                    'body' => $body
                ]);
            }
        }
    }
}

这里是发送邮件的代码

<?php

namespace app\common\service\email;

use app\common\service\ConfigService;
use PHPMailer\PHPMailer\PHPMailer;
use PHPMailer\PHPMailer\SMTP;
use think\Exception;

class EmailService
{
    private static $instance = null;
    private static $error = null;
    private $mailer;

    public function __construct()
    {
        $this->mailer = new PHPMailer(true);
        $this->mailer->SMTPDebug = SMTP::DEBUG_OFF;
        $this->mailer->isSMTP();
        $this->mailer->Host = ConfigService::get('email', 'email_service');
        $this->mailer->SMTPAuth = true;
        $this->mailer->Username = ConfigService::get('email', 'email_username');
        $this->mailer->Password = ConfigService::get('email', 'email_password');
        $this->mailer->SMTPSecure = PHPMailer::ENCRYPTION_SMTPS;
        $this->mailer->Port = ConfigService::get('email', 'email_port');
        $this->mailer->setFrom(ConfigService::get('email', 'email_send_email'), 'JCL Mailer');
        $this->mailer->isHTML(true);
    }

    public static function init()
    {
        if (self::$instance === null) {
            self::$instance = new self();
        }
        return self::$instance;
    }

    public static function getError()
    {
        return self::$error;
    }

    public function sendRegister($email, $params = [])
    {
        try {
            $this->mailer->addAddress($email, 'User');
            $this->mailer->Subject = $params['subject'] ?? 'Default Subject';
            $this->mailer->Body = $params['body'] ?? 'Default Body';
            $this->mailer->send();
            return true;
        } catch (Exception $e) {
            self::$error = $e->getMessage();
            return false;
        }
    }

    // 发送新报价提醒  待最后来统计通知类型再来合并处理

    public function sendNewOfferNoticeToAdmin($email, $params = [])
    {
        try {
            $this->mailer->addAddress($email, 'Site Notice');
            $this->mailer->Subject = $params['subject'] ?? 'Default Subject';
            $this->mailer->Body = $params['body'] ?? 'Default Body';
            $this->mailer->CharSet = 'UTF-8';
            $this->mailer->send();
            return true;
        } catch (Exception $e) {
            self::$error = $e->getMessage();
            return false;
        }
    }

    /**
     * 封装公共邮件发送,数据接口如下
     * @param $data
     * @return void
     * @throws \PHPMailer\PHPMailer\Exception
     *
     *  $data=[
     *     'emails'=>'',  //这个是接收邮件的人
     *     'subject'  =>'',
     *     'body'  =>''
     * ]
     */
    public function sendEmail($data)
    {
        //注意这里不能使用thinkphp的查询
        try {
            $this->mailer->addAddress($data['email'], 'User');
            $this->mailer->Subject = $data['subject'] ?? 'Default Subject';
            $this->mailer->Body = $data['body'] ?? 'Default Body';
            $this->mailer->send();
            //Log::info("邮件发送成功:".$email);
            echo "邮件发送成功:" . $data['email'];
        } catch (Exception $e) {
            // self::$error = $e->getMessage();
            //Log::error("邮件发送失败:".$e->getMessage());
            echo "邮件发送失败:" . $e->getMessage();
        }

    }
}