[大数据]tp6使用Kafka进行发邮件Excel

使用前需要先进行Kafka的配置和Java JDK的安装

参考文档

然后在框架中安装Excel

composer require phpmailer/phpmailer 

然后在配置文件中进行公共参数的设置

//邮件发送配置
    'email' => [
        'CharSet' => 'UTF-8',//编码格式为utf8
        'Host' => 'smtp.qq.com',//SMTP服务器 以QQ邮箱为例子
        'SMTPAuth' => true,//启用SMTP认证,是否使用身份验证
        'Username' => '1919814727@qq.com',//发送方的邮箱
        'Password' => 'izhbczgidupvcfci',//你的授权码
        'SMTPSecure' => 'ssl',// 设置安全验证方式为ssl
        'Port' => '465',//邮件发送端口
    ],

然后封装公共方法

if(!function_exists('sendEmail')) {
    /**
     * @param $toemails
     * 收件人邮箱
     * @throws \PHPMailer\PHPMailer\Exception
     */
    function sendMail($toemails)
    {
        $toemail = $toemails;//定义收件人的邮箱
        $mail = new PHPMailer\PHPMailer\PHPMailer();
        $mail->isSMTP();// 使用SMTP服务
        $mail->CharSet = \think\facade\Config::get('app.email.CharSet');// 编码格式为utf8，不设置编码的话，中文会出现乱码
        $mail->Host = \think\facade\Config::get('app.email.Host');// 发送方的SMTP服务器地址
        $mail->SMTPAuth = \think\facade\Config::get('app.email.SMTPAuth');// 是否使用身份验证
        $mail->Username = \think\facade\Config::get('app.email.Username');// 发送方的163邮箱用户名，就是你申请163的SMTP服务使用的163邮箱</span><span style="color:#333333;">
        $mail->Password = \think\facade\Config::get('app.email.Password');// 发送方的邮箱密码，注意用163邮箱这里填写的是“客户端授权密码”而不是邮箱的登录密码！</span><span style="color:#333333;">
        $mail->SMTPSecure = \think\facade\Config::get('app.email.SMTPSecure');// 使用ssl协议方式</span><span style="color:#333333;">
        $mail->Port = \think\facade\Config::get('app.email.Port');// 163邮箱的ssl协议方式端口号是465/994
        $mail->setFrom("1919814727@qq.com", "Mailer");// 设置发件人信息，如邮件格式说明中的发件人，这里会显示为Mailer(xxxx@163.com），Mailer是当做名字显示
        $mail->addAddress($toemail, 'YangShuBin');// 设置收件人信息，如邮件格式说明中的收件人，这里会显示为Liang(yyyy@163.com)
        $mail->addReplyTo("1919814727@qq.com", "WangXiaoQiang");// 设置回复人信息，指的是收件人收到邮件后，如果要回复，回复邮件将发送到的邮箱地址
        $mail->Subject = "这是一个测试邮件";// 邮件标题
        $mail->Body = "邮件内容是 <b>您的验证码是：123456</b>，哈哈哈！";// 邮件正文
        if (!$mail->send()) {// 发送邮件
            echo "Message could not be sent.";
            echo "Mailer Error: " . $mail->ErrorInfo;// 输出错误信息
        } else {
            echo '发送成功';
        }
    }
}

启动kafka

输入以下命令

bin/windows/zookeeper-server-start.bat config/zookeeper.properties

再打开一个命令框

bin/windows/kafka-server-start.bat config/server.properties

添加生产者

  //设置发送邮件的生产者producer
    public function sendEmail()
    {
        $rk = new \RdKafka\Producer();
        $rk->addBrokers("127.0.0.1:9092"); //kafka服务器地址
        // $rk->addBrokers("10.0.0.1,10.0.0.2");  //多服务器地址

        $topic = $rk->newTopic("wxq"); //topic名称
        $toemails = [
            'YSB'=>'1sfds015@qq.com',
            'TYS'=>'g.ysdfsan.s@qq.com',
            'TYS1'=>'30473ssdfsf29@qq.com',
            'YLY'=>'39sdff28sdfsdf7213@qq.com',
            'LGF'=>'84fsd9127@qq.com',
        ];
        foreach($toemails as $k=>$v) {
            $topic->produce(RD_KAFKA_PARTITION_UA, 0, $this->sendMails($v));
            $rk->poll(0);
            sleep(5);
        }
        while ($rk->getOutQLen() > 0) {
            $rk->poll(50);
        }
    }

        //掉哟弄个发送邮件方法
        public function sendMails($toemails)
        {
            //$toemails = '1710150890@qq.com';//定义收件人的邮箱
            sendMail($toemails);
        }

<?php

$conf = new \RdKafka\Conf();

$conf->set('group.id', 'myConsumerGroup');

$rk = new \RdKafka\Consumer($conf);
$rk->addBrokers("127.0.0.1:9092");

$topicConf = new \RdKafka\TopicConf();
$topicConf->set('auto.commit.interval.ms', 100);
$topicConf->set('offset.store.method', 'file');
$topicConf->set('offset.store.path', sys_get_temp_dir());
$topicConf->set('auto.offset.reset', 'smallest');

$topic = $rk->newTopic("wxq", $topicConf);

// Start consuming partition 0
$topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

while (true) {
    $message = $topic->consume(0, 1 * 10000);
    switch ($message->err) {
        case RD_KAFKA_RESP_ERR_NO_ERROR:
            //没有错误打印信息
            var_dump($message);
            break;
        case RD_KAFKA_RESP_ERR__PARTITION_EOF:
            echo "等待接收信息\n";
            break;
        case RD_KAFKA_RESP_ERR__TIMED_OUT:
            echo "超时\n";
            break;
        default:
            throw new \Exception($message->errstr(), $message->err);
            break;
    }
}

创建消费者（这个放在原生php文件中就可以，需要使用cli运行模式来运行，否则出不来效果）