es_producer/app/service/DataSyncService.php

<?php
namespace App\Service;

use Hyperf\DbConnection\Db;
use Hyperf\Elasticsearch\ClientBuilderFactory;
use Elasticsearch\Client;

class DataSyncService
{
    protected $client;

    public function __construct(ClientBuilderFactory $clientBuilderFactory)
    {
        $this->client = $clientBuilderFactory->create();
    }

    public function syncArticles()
    {
        $articles = Db::table('article')->get();
        $articleData = Db::table('article_data')->get();

        foreach ($articles as $article) {
            $data = $articleData->firstWhere('article_id', $article->id);
            if ($data) {
                $params = [
                    'index' => 'articles',
                    'id' => $article->id,
                    'body' => [
                        'id' => $article->id,
                        'catid' => $article->catid,
                        'level' => $article->level,
                        'title' => $article->title,
                        'introduce' => $article->introduce,
                        'tag' => $article->tag,
                        'keyword' => $article->keyword,
                        'author' => $article->author,
                        'copyfrom' => $article->copyfrom,
                        'fromurl' => $article->fromurl,
                        'hits' => $article->hits,
                        'ip' => $article->ip,
                        'status' => $article->status,
                        'islink' => $article->islink,
                        'linkurl' => $article->linkurl,
                        'imgurl' => $article->imgurl,
                        'admin_user_id' => $article->admin_user_id,
                        'cat_arr_id' => $article->cat_arr_id,
                        'created_at' => $article->created_at,
                        'updated_at' => $article->updated_at,
                        'is_original' => $article->is_original,
                        'content' => $data->content,
                    ],
                ];

                try {
                    $this->client->index($params);
                } catch (\Exception $e) {
                    // 记录日志或处理异常
                    // \Hyperf\Utils\LoggerFactory::getInstance()->error($e->getMessage());
                }
            }
        }
    }
}