csharp Задачадлямногопотокасблокировкойтаблиц

Posted 2021-05-12

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading;
using System.IO;
using System.Text.RegularExpressions;
using ZennoLab.CommandCenter;
using ZennoLab.InterfacesLibrary;
using ZennoLab.InterfacesLibrary.ProjectModel;
using ZennoLab.InterfacesLibrary.ProjectModel.Collections;
using ZennoLab.InterfacesLibrary.ProjectModel.Enums;
using ZennoLab.Macros;
using Global.ZennoExtensions;
using ZennoLab.Emulation;

namespace ZennoLab.OwnCode
{
    /// <summary>
	/// A simple class of the common code
	/// </summary>
	public class CommonCode
	{
		/// <summary>
		/// Lock this object to mark part of code for single thread execution
		/// </summary>
		public static object SyncObject = new object();

		// Insert your code here
	}
}

//Данный сниппет предназначен для демонстрации многопоточной работы с БД (добавьте текущий проект в ZennoPoster и запустите в несколько потоков)

string strDBConnString = project.Variables["cfg_db_connstring"].Value;
int intMaxTasks = Convert.ToInt32(project.Variables["cfg_tasks_per_execution"].Value);
bool blnLockTables = Convert.ToBoolean(project.Variables["cfg_use_table_locks"].Value);

//добавляем несколько "заданий" в таблицу db_jobs (при штатной работе фермы эта информация появлялась бы в таблице в результате других процессов)
List<int> lstRandomProfiles = new List<int>() {1, 2, 3, 4, 5};
lstRandomProfiles.Shuffle();
string strAddTasksQuery = String.Format("INSERT INTO db_tasks (profile_id, task_type, post_text, post_image, object) VALUES ({0}, 'post', 'this is a texxt', 'image1.jpg', 'http://vk.com/supergruppa');" +
    "INSERT INTO db_tasks (profile_id, task_type, object) VALUES ({1}, 'like', 'https://vk.com/wall-2158488_268002');" +
	"INSERT INTO db_tasks (profile_id, task_type, object) VALUES ({2}, 'invite', 'https://vk.com/id139687748');" +
	"INSERT INTO db_tasks (profile_id, task_type, object) VALUES ({3}, 'like', 'https://vk.com/wall-2158488_636329');" +
	"INSERT INTO db_tasks (profile_id, task_type, post_text, post_image, object) VALUES ({4}, 'post', 'text namba tu', 'image2.jpg', 'http://vk.com/goodgruppa');", 
	lstRandomProfiles[0], lstRandomProfiles[1], lstRandomProfiles[2], lstRandomProfiles[3], lstRandomProfiles[4]);
ZennoPoster.Db.ExecuteNonQuery(strAddTasksQuery, null, ZennoLab.InterfacesLibrary.Enums.Db.DbProvider.MySqlClient, strDBConnString);

//получаем свободный профиль (со статусом ready), для которого количество "заданий" больше нуля ("нативная" работа с MySqlClient)
var ProfileSelectCommand = new MySql.Data.MySqlClient.MySqlCommand();
try{
	ProfileSelectCommand.Connection = new MySql.Data.MySqlClient.MySqlConnection(strDBConnString);
}catch(Exception e){
	return "MySql connect failed";
}
ProfileSelectCommand.Connection.Open();

//блокируем таблицу db_profiles для записи другими потоками
string strLockQuery = "LOCK TABLES db_profiles WRITE, db_tasks READ";
MySql.Data.MySqlClient.MySqlCommand NonQueryDBCommand = new MySql.Data.MySqlClient.MySqlCommand(strLockQuery, ProfileSelectCommand.Connection);
if (blnLockTables) NonQueryDBCommand.ExecuteNonQuery();

string strSelectProfileQuery = "SELECT db_profiles.id, login, password, db_profiles.status " +
	"FROM db_profiles LEFT JOIN db_tasks ON db_profiles.id = db_tasks.profile_id " +
	"WHERE db_tasks.status='wait' " +
	"GROUP BY login, password, db_profiles.status " +
	"HAVING db_profiles.status='ready' AND COUNT(task_type)>0 " +
	"ORDER BY COUNT(db_tasks.task_type) DESC LIMIT 1;";
ProfileSelectCommand.CommandText=strSelectProfileQuery;
var ProfileLoadReader = ProfileSelectCommand.ExecuteReader();

string strCurrProfileID=String.Empty;
string strLogin = String.Empty;
string strPassword = String.Empty;
string strCurrProfileStatus = String.Empty;
if(ProfileLoadReader.HasRows) {
	//считываем данные из запроса
	ProfileLoadReader.Read();
	strCurrProfileID=ProfileLoadReader["id"].ToString();
	strLogin = ProfileLoadReader["login"].ToString();
	strPassword = ProfileLoadReader["password"].ToString();
	strCurrProfileStatus = ProfileLoadReader["status"].ToString();
	project.SendInfoToLog(String.Format("Получили профиль с id: {0} ({1})", strCurrProfileID, strLogin), true);
}else{
	project.SendInfoToLog("Нет ожидающих профилей с заданиями. Завершаем работу", true);
}
ProfileLoadReader.Close();

if (strCurrProfileID!=String.Empty) {
	//помечаем профиль как "занятый" (busy)
	NonQueryDBCommand.CommandText=String.Format("UPDATE db_profiles SET status='busy', last_used=NOW() WHERE id={0};", strCurrProfileID);
	NonQueryDBCommand.ExecuteNonQuery();
}

if (blnLockTables) {
	//снимаем блокировку таблиц
	NonQueryDBCommand.CommandText="UNLOCK TABLES;";
	NonQueryDBCommand.ExecuteNonQuery();
}

//закрываем подключение к БД
ProfileSelectCommand.Connection.Close();

if (strCurrProfileID==String.Empty) return "noprofiles";

//сохраняем id текущего профиля в переменной уровня проекта на случай возникновения ошибки (в этом случае по BadEnd разлочим профиль)
project.Variables["str_current_profile"].Value = strCurrProfileID;

//цикл "выполнения заданий"
int intCompletedTasks = 0; //количество выполненных заданий
do {
	//получаем задание для выполнения
	string strTasksQuery = String.Format("SELECT id, task_type, post_text, post_image, object FROM db_tasks WHERE profile_id={0} AND status='wait' LIMIT 1;", strCurrProfileID);
	string strTask = ZennoPoster.Db.ExecuteQuery(strTasksQuery, null, ZennoLab.InterfacesLibrary.Enums.Db.DbProvider.MySqlClient, strDBConnString, "|", Environment.NewLine);
	if (strTask.Length==0) {
		project.SendInfoToLog(String.Format("Для профиля {0} ({1}) успешно выполнено {2} заданий. Больше в БД заданий для профиля нет, завершаем работу.", strCurrProfileID, strLogin, intCompletedTasks), true);
		break;
	}
	string[] arrTask = strTask.Split('|'); //обратите внимание, в данных таблицы db_tasks не должно быть символов |
	
	//получаем параметры текущего задания
	string strTaskId = arrTask[0];
	string strWhatToDo = arrTask[1];
	string strText = arrTask[2];
	string strImage = arrTask[3];
	string strTaskObject = arrTask[4];
	
	project.SendInfoToLog(String.Format("Выполняем задание id {0} ({1}) для профиля с id {2} ({3})", strTaskId, strWhatToDo, strCurrProfileID, strLogin), true);
	
	//запускаем выполнение задания
	string strTaskResultMessage = String.Empty;
	switch(strWhatToDo) {
		case "post":
			strTaskResultMessage = project.Context["fnPost"](strTaskObject, strText, strImage);
			break;
		case "like":
			strTaskResultMessage = project.Context["fnLike"](strTaskObject);
			break;
		case "invite":
			strTaskResultMessage = project.Context["fnInvite"](strTaskObject);
			break;
	}
	
	//анализируем результат выполнение задания
	string strTaskResult = String.Empty;
	if (strTaskResultMessage.StartsWith("ok:")){
		strTaskResult = "success";
	}else{
		strTaskResult = "error";
	}
	//помечаем задачу выполненной, сохраняем результат выполнения
	ZennoPoster.Db.ExecuteNonQuery(String.Format("UPDATE db_tasks SET status='{0}', status_message='{1}', change_dt=NOW() WHERE id={2};", strTaskResult, strTaskResultMessage, strTaskId), null, ZennoLab.InterfacesLibrary.Enums.Db.DbProvider.MySqlClient, strDBConnString);
	
	if (strTaskResultMessage.Contains("заблокирован")) {
		project.SendWarningToLog(String.Format("Профиль с id {0} ({1}) заблокирован. Завершаем работу", strCurrProfileID, strLogin), true);
		//отменяем все задания в базе для данного профиля (в реальности можно переназначать задания на "любой" профиль)
		ZennoPoster.Db.ExecuteNonQuery(String.Format("UPDATE db_tasks SET status='cancelled', status_message='profile blocked', change_dt=NOW() WHERE profile_id={0} AND status='wait';", strCurrProfileID), null, ZennoLab.InterfacesLibrary.Enums.Db.DbProvider.MySqlClient, strDBConnString);
		strCurrProfileStatus = "blocked"; //меняем статус профиля в переменной на blocked
		break; //прерываем цикл выполнения заданий
	}
	
	intCompletedTasks++;
	
} while (intCompletedTasks<intMaxTasks);

//помечаем профиль как "свободный" (ready). блокировка таблицы при этом не требуется
ZennoPoster.Db.ExecuteNonQuery(String.Format("UPDATE db_profiles SET status='{0}', last_used=NOW() WHERE id={1};", strCurrProfileStatus, strCurrProfileID), null, ZennoLab.InterfacesLibrary.Enums.Db.DbProvider.MySqlClient, strDBConnString);

return "finished";

//ПРИМЕЧАНИЯ:
//1.