MySQL服务器在运行Laravel Queue时已经消失了
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了MySQL服务器在运行Laravel Queue时已经消失了相关的知识，希望对你有一定的参考价值。
我正在尝试使用laravel队列将数据从CSV文件导入数据库。这些CSV文件非常庞大，行数约为500k。
我已经了解到使用laravel队列我们不需要考虑连接超时，但这看起来并不正确。也许我错了。
如果这些方法有任何问题，请检查我的工作代码。我使用“League Csv”来读取CSV文件。
public function __construct($data,$error_arr,$error_row_numbers) {
$this->data = $data;
$this->error_arr = $error_arr;
$this->error_row_numbers = $error_row_numbers;
}
/**
* Execute the job.
*
* @return void
*/
public function handle()
{
$offset = $this->data['offset'];
$limit = $this->data['limit'];
$filename = $this->data['file_name'];
$service = new Service();
$table = 'committees';
$dbase = new Committee();
//map_data have array about which csv column
//should be inserted in which column of database
$map_data = $this->data['map_data'];
//get all columns name of a table
$db_header_obj = new Committee();
$db_header = $db_header_obj->getTableColumns();
$csv_file_path = storage_path('app/files/committee/').$filename;
if (!ini_get("auto_detect_line_endings")) {
ini_set("auto_detect_line_endings", TRUE);
}
$csv = Reader::createFromPath($csv_file_path, 'r');
$csv->setOutputBOM(Reader::BOM_UTF8);
$csv->addStreamFilter('convert.iconv.ISO-8859-15/UTF-8');
$csv->setHeaderOffset(0);
$csv_header = $csv->getHeader();
$rec_arr = array();
$records = array();
$records_arr = array();
$stmt = (new Statement())
->offset($offset)
->limit($limit)
;
$records = $stmt->process($csv);
foreach ($records as $record)
{
$rec_arr[] = array_values($record);
}
//trim index if the value of an array is empty
$records_arr = $service->trimArray($rec_arr);
if(count($records_arr)>0)
{
foreach($records_arr as $ck => $cv){
$committee_arr = array();
foreach ($map_data as $mk => $mv) {
if(isset($mv)){
$data_type = $service->getDatabaseColumnType($table,$mv);
//if data is one of datetime data type
//then format the csv data to mysql datetime format
if($data_type == 'date' || $data_type == 'datetime' || $data_type == 'timestamp'){
$datetime = (array)$cv[$mk];
$dt = array_shift($datetime);
$dt = date('Y-m-d h:i:s', strtotime($dt));
$committee_arr[$mv] = $dt;
}else{
$committee_arr[$mv] = $cv[$mk];
}
}
}
$error_encountered = false;
DB::beginTransaction();
if(!empty($committee_arr['com_id'])){
try{
$committee_row = Committee::updateOrCreate(
['com_id' => $committee_arr['com_id']],
$committee_arr
);
if ($committee_row->wasRecentlyCreated === true) {
$committee_row->created_by = $this->data['user_id'];
}else{
$committee_row->updated_by = $this->data['user_id'];
}
$committee_row->save();
} catch (Exception $e) {
$error_encountered = true;
$this->error_arr[] = $e->getMessage();
$this->error_row_numbers[] = $this->data['row_value'];
}
}
DB::commit();
//just to keep track which row is currently processing
//so that user can be notified in which row of csv
//there is an error
$this->data['row_value'] = $this->data['row_value'] + 1;
}
//offset just to start fectch next chunk of data from csv
$this->data['offset'] = $offset + $limit;
//Call to same job but with increased offset value
$committeeInsertJob = (new StoreCommittee($this->data,$this->error_arr,$this->error_row_numbers))->delay(Carbon::now()->addSeconds(3));
dispatch($committeeInsertJob);
}else{
//Store activity just to keep track of activity
$activity = new Activity();
$activity->url = $this->data['url'];
$activity->action = 'store';
$activity->description = $table;
$activity->user_id = $this->data['user_id'];
$activity->created_at = date('Y-m-d H:i:s');
$activity->save();
$arr_data = [
'filename' => $filename,
'user_name' => $this->data['user_name'],
'error' => $this->error_arr,
'error_row_numbers' => $this->error_row_numbers
];
//Notify user that the job is complete
Mail::to($this->data['user_email'])->send(new CSVImportJobComplete($arr_data));
}
if (!ini_get("auto_detect_line_endings")) {
ini_set("auto_detect_line_endings", FALSE);
}
}
错误：从（laravel.log内部存储）
[2019-04-05 07:13:23] local.ERROR: PDOStatement::execute(): MySQL server has gone away (SQL: insert into `jobs` (`queue`, `attempts`, `reserved_at`, `available_at`, `created_at`, `payload`) values (default, 0, , 1554448406, 1554448403, ....................................................(long list)
来自：命令终端
$ php artisan queue:work --tries=3
[2019-04-05 07:09:11][1] Processing: AppJobsStoreCommittee
[2019-04-05 07:09:33][1] Processed: AppJobsStoreCommittee
[2019-04-05 07:09:36][2] Processing: AppJobsStoreCommittee
[2019-04-05 07:09:58][2] Processed: AppJobsStoreCommittee
[2019-04-05 07:10:01][3] Processing: AppJobsStoreCommittee
[2019-04-05 07:10:23][3] Processed: AppJobsStoreCommittee
[2019-04-05 07:10:26][4] Processing: AppJobsStoreCommittee
[2019-04-05 07:10:48][4] Processed: AppJobsStoreCommittee
[2019-04-05 07:10:51][5] Processing: AppJobsStoreCommittee
[2019-04-05 07:11:13][5] Processed: AppJobsStoreCommittee
[2019-04-05 07:11:17][6] Processing: AppJobsStoreCommittee
[2019-04-05 07:11:40][6] Processed: AppJobsStoreCommittee
[2019-04-05 07:11:43][7] Processing: AppJobsStoreCommittee
[2019-04-05 07:12:05][7] Processed: AppJobsStoreCommittee
[2019-04-05 07:12:08][8] Processing: AppJobsStoreCommittee
[2019-04-05 07:12:31][8] Processed: AppJobsStoreCommittee
[2019-04-05 07:12:34][9] Processing: AppJobsStoreCommittee
[2019-04-05 07:12:57][9] Processed: AppJobsStoreCommittee
[2019-04-05 07:13:00][10] Processing: AppJobsStoreCommittee
dell@DESKTOP-UQ2 MINGW64 /d/wamp64/www/project(master)
$
(it stops without any error or failed notifications)
有什么我可以改善我的工作逻辑吗？我如何处理所有这些连接丢失，或超时超时或其他一些东西？我不认为增加超时是解决方案。因为不能保证它会在这个固定的时间内完成。
相反，有没有办法，连接可以关闭，并再次重新连接每个队列工作？
答案
试图解决方案
您解析了CSV文件，并尝试通过单个查询发送整个内容。 MySQL包含的变量阻止它接受过大的查询。它被称为max_allowed_packet
你这样做的原因是表现。但是，在处理关于数据量过大的查询时，您可以使用与网络/ MySQL相关的许多变量中的一个。
改进了解决方案标
- 清理查询，以便可以看到正在完成的工作
- 快速查询，因此可以发送大量数据以快速进行写入
- 不要达到限制变量的值，例如max_packet_size
解
- 准确地准备了一次声明。准备好的语句使用一次，执行多次
- 解析CSV并循环记录
- 将值绑定到预准备语句并在循环中执行它
- 为了使一切更快，请使用事务。在事务中包装每1000条记录。这将让你编写简单的插入查询，但它们会很快，因为MySQL将复用写入
你正在使用Laravel，所以上面的步骤非常容易
使用laravel的伪代码
$csv = collect([]); // This is the array holding your CSV records
// Split the array into chunks. Let's assume you want to insert 1000 records in one attempt
$chunk_count = ceil($csv->count() / 1000);
$csv->chunk($chunk_count)->map(function($chunk) {
DB::beginTransaction();
// Create a record
$chunk->map(function($data) {
StoreCommittee::create($data);
});
DB::commit();
});
以上是关于MySQL服务器在运行Laravel Queue时已经消失了的主要内容，如果未能解决你的问题，请参考以下文章
Laravel queue:listen 一直占用服务器所有内存
Laravel 运行队列处理器 queue:work 与 queue:listen 区别及 Windows 终端命令问题