問題描述

再使用workerman/redis-queue循環(huán)發(fā)送郵件，比如有4個(gè)郵件，在subscribe里是能打印四個(gè)郵件，但是每個(gè)郵箱會(huì)收到四條一模一樣的郵件，按道理的是話每個(gè)郵箱只應(yīng)該收到一條郵件才對(duì),請(qǐng)問這個(gè)要如何處理？謝謝

程序代碼或配置

<?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, '守護(hù)進(jìn)程')
            ->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());
                }
            });
            // 消費(fèi)失敗觸發(fā)的回調(diào)(可選)
            $client->onConsumeFailure(function (\Throwable $exception, $package) {
                //\think\facade\Log::log("隊(duì)列 " . $package['queue'] . " 消費(fèi)失敗:".$exception->getMessage());
                var_export("隊(duì)列 " . $package['queue'] . " 消費(fèi)失敗:" . $exception->getMessage());
            });
        };

        Worker::runAll();
    }

}

重現(xiàn)問題的步驟

    //這里調(diào)用發(fā)送
  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) {
            //如果是注冊(cè)
            case 'register':
                $subject = $params['name'] . '新用戶注冊(cè)郵箱提醒';
                $body = "<div>
                <div><img style='width: 50px;height: 40px;' src='" . $params['logo'] . "'/></div>
                <h4>新用戶注冊(cè)信息</h4>
                <p>用戶名：" . $params['account'] . "</p>
                <p>郵箱：" . $params['emails'][0] . "</p>
                </div>";
                break;
            case 'companyAuthAfter':
                $subject = '新增企業(yè)認(rèn)證';
                $body = "<div>
                <div><img style='width: 50px;height: 40px;' src='" . $siteLogo . "'/></div>
                <h4>企業(yè)已完成認(rèn)證，請(qǐng)進(jìn)入平臺(tái)查看相關(guān)信息</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
                ]);
            }
        }
    }
}

這里是發(fā)送郵件的代碼

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

    // 發(fā)送新報(bào)價(jià)提醒  待最后來統(tǒng)計(jì)通知類型再來合并處理

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

    /**
     * 封裝公共郵件發(fā)送,數(shù)據(jù)接口如下
     * @param $data
     * @return void
     * @throws \PHPMailer\PHPMailer\Exception
     *
     *  $data=[
     *     'emails'=>'',  //這個(gè)是接收郵件的人
     *     '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("郵件發(fā)送成功:".$email);
            echo "郵件發(fā)送成功:" . $data['email'];
        } catch (Exception $e) {
            // self::$error = $e->getMessage();
            //Log::error("郵件發(fā)送失敗:".$e->getMessage());
            echo "郵件發(fā)送失敗:" . $e->getMessage();
        }

    }
}

操作系統(tǒng)環(huán)境及workerman/webman等具體版本

linux