es_producer/app/Command/SyncArticlesCommand.php

<?php
declare (strict_types = 1);

namespace App\Command;

use Hyperf\Command\Annotation\Command;
use Swoole\Coroutine;
use Hyperf\DbConnection\Db;
use Hyperf\Elasticsearch\ClientBuilderFactory;
use Psr\Container\ContainerInterface;
use App\Service\DataSyncService;
use Hyperf\Command\Command as HyperfCommand;
use App\Service\ElasticsearchService;

/**
 * @Command
 */
class SyncArticlesCommand extends HyperfCommand
{
    protected $dataSyncService;
    protected $elasticsearchService;

    public function __construct(ContainerInterface $container)
    {
        parent::__construct('sync:articles');
        $this->dataSyncService = $container->get(DataSyncService::class);
        $this->elasticsearchService = $container->get(ElasticsearchService::class);
    }

    public function configure()
    {
        parent::configure();
        $this->setName('sync:articles')
            ->setDescription('Synchronize articles to Elasticsearch');
    }

    public function handle()
    {
        Coroutine::create(function () {
            $this->line('========================= started.');

            // 检查索引是否存在
            $indexExists = $this->elasticsearchService->indexExists('articles');
            if (!$indexExists) {
                // 如果索引不存在，则创建索引
                try {
                    $response = $this->elasticsearchService->createIndex();
                    $this->line('Index created successfully: ' . json_encode($response));
                } catch (\Exception $e) {
                    $this->error('Failed to create index: ' . $e->getMessage());
                    return;
                }
            } else {
                $this->line('Index already exists.');
            }

            // 同步数据
            try {
                $this->dataSyncService->syncArticles();
                $this->line('Articles synchronization completed.');
            } catch (\Exception $e) {
                $this->error('Failed to synchronize articles: ' . $e->getMessage());
            }
        });
        $this->line('Articles synchronization started.');
    }
}