Because of deadlock and performance issues, mysql does not use transactions when consuming messages, and requeue the message to queue table when they exception. (#36,#68)

Author: yang-xiaodong

src/DotNetCore.CAP.MySql/MySqlFetchedMessage.cs

@@ -1,29 +1,19 @@
-using System;
-using System.Data;
-using System.Threading;
-using Dapper;
+using Dapper;
 using DotNetCore.CAP.Models;
+using MySql.Data.MySqlClient;
 
 namespace DotNetCore.CAP.MySql
 {
     public class MySqlFetchedMessage : IFetchedMessage
     {
-        private static readonly TimeSpan KeepAliveInterval = TimeSpan.FromMinutes(1);
-        private readonly IDbConnection _connection;
-        private readonly object _lockObject = new object();
-        private readonly Timer _timer;
-        private readonly IDbTransaction _transaction;
+        private readonly string _connectionString = null;
 
-        public MySqlFetchedMessage(int messageId,
-            MessageType type,
-            IDbConnection connection,
-            IDbTransaction transaction)
+        public MySqlFetchedMessage(int messageId, MessageType type, string connectionString)
         {
             MessageId = messageId;
             MessageType = type;
-            _connection = connection;
-            _transaction = transaction;
-            _timer = new Timer(ExecuteKeepAliveQuery, null, KeepAliveInterval, KeepAliveInterval);
+
+            _connectionString = connectionString;
         }
 
         public int MessageId { get; }
@@ -32,43 +22,21 @@ public MySqlFetchedMessage(int messageId,
 
         public void RemoveFromQueue()
         {
-            lock (_lockObject)
-            {
-                _transaction.Commit();
-            }
+            // ignored
         }
 
         public void Requeue()
         {
-            lock (_lockObject)
+            using (var connection = new MySqlConnection(_connectionString))
             {
-                _transaction.Rollback();
+                connection.Execute("insert into `cap.queue`(`MessageId`,`MessageType`) values(@MessageId,@MessageType);"
+                    , new {MessageId, MessageType });
             }
         }
 
         public void Dispose()
         {
-            lock (_lockObject)
-            {
-                _timer?.Dispose();
-                _transaction.Dispose();
-                _connection.Dispose();
-            }
-        }
-
-        private void ExecuteKeepAliveQuery(object obj)
-        {
-            lock (_lockObject)
-            {
-                try
-                {
-                    _connection?.Execute("SELECT 1", _transaction);
-                }
-                catch
-                {
-                    // ignored
-                }
-            }
+            // ignored
         }
     }
 }

src/DotNetCore.CAP.MySql/MySqlStorage.cs

@@ -53,7 +53,8 @@ protected virtual string CreateDbTablesScript(string prefix)
                 $@"
 CREATE TABLE IF NOT EXISTS `{prefix}.queue` (
   `MessageId` int(11) NOT NULL,
-  `MessageType` tinyint(4) NOT NULL
+  `MessageType` tinyint(4) NOT NULL,
+  `ProcessId` varchar(50) DEFAULT NULL
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
 CREATE TABLE IF NOT EXISTS `{prefix}.received` (

src/DotNetCore.CAP.MySql/MySqlStorageConnection.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Data;
 using System.Threading.Tasks;
 using Dapper;
 using DotNetCore.CAP.Infrastructure;
@@ -43,15 +42,11 @@ public async Task<CapPublishedMessage> GetPublishedMessageAsync(int id)
         public Task<IFetchedMessage> FetchNextMessageAsync()
         {
             var sql = $@"
-SELECT `MessageId`,`MessageType` FROM `{_prefix}.queue` LIMIT 1 FOR UPDATE;
-DELETE FROM `{_prefix}.queue` LIMIT 1;";
+UPDATE `{_prefix}.queue` SET `ProcessId`=@ProcessId WHERE `ProcessId` IS NULL LIMIT 1;
+SELECT `MessageId`,`MessageType` FROM `{_prefix}.queue` WHERE `ProcessId`=@ProcessId;
+DELETE FROM `{_prefix}.queue` WHERE `ProcessId`=@ProcessId";
 
-            // The following `sql` can improve performance, but repeated consumption occurs in multiple instances
-            //var sql = $@"
-            //SELECT @MId:=`MessageId` as MessageId, @MType:=`MessageType` as MessageType FROM `{_prefix}.queue` LIMIT 1;
-            //DELETE FROM `{_prefix}.queue` where `MessageId` = @MId AND `MessageType`=@MType;";
-
-            return FetchNextMessageCoreAsync(sql);
+            return FetchNextMessageCoreAsync(sql, new { ProcessId = Guid.NewGuid().ToString() });
         }
 
         public async Task<CapPublishedMessage> GetNextPublishedMessageToBeEnqueuedAsync()
@@ -122,11 +117,6 @@ public async Task<IEnumerable<CapReceivedMessage>> GetFailedReceivedMessages()
             }
         }
 
-
-        public void Dispose()
-        {
-        }
-
         public bool ChangePublishedState(int messageId, string state)
         {
             var sql =
@@ -151,44 +141,20 @@ public bool ChangeReceivedState(int messageId, string state)
 
         private async Task<IFetchedMessage> FetchNextMessageCoreAsync(string sql, object args = null)
         {
-            //here don't use `using` to dispose
-            var connection = new MySqlConnection(Options.ConnectionString);
-            await connection.OpenAsync();
-            var transaction = connection.BeginTransaction(IsolationLevel.ReadCommitted);
-            FetchedMessage fetchedMessage = null;
-            try
-            {
-                //fetchedMessage = await connection.QuerySingleOrDefaultAsync<FetchedMessage>(sql, args, transaction);
-                // An anomaly with unknown causes, sometimes QuerySingleOrDefaultAsync can't return expected result.
-                using (var reader = connection.ExecuteReader(sql, args, transaction))
-                {
-                    while (reader.Read())
-                    {
-                        fetchedMessage = new FetchedMessage
-                        {
-                            MessageId = (int)reader.GetInt64(0),
-                            MessageType = (MessageType)reader.GetInt64(1)
-                        };
-                    }
-                }
-            }
-            catch (MySqlException)
+            FetchedMessage fetchedMessage;
+            using (var connection = new MySqlConnection(Options.ConnectionString))
             {
-                transaction.Dispose();
-                connection.Dispose();
-                throw;
+                fetchedMessage = await connection.QuerySingleOrDefaultAsync<FetchedMessage>(sql, args);
             }
 
             if (fetchedMessage == null)
-            {
-                transaction.Rollback();
-                transaction.Dispose();
-                connection.Dispose();
                 return null;
-            }
 
-            return new MySqlFetchedMessage(fetchedMessage.MessageId, fetchedMessage.MessageType, connection,
-                transaction);
+            return new MySqlFetchedMessage(fetchedMessage.MessageId, fetchedMessage.MessageType, Options.ConnectionString);
+        }
+
+        public void Dispose()
+        {
         }
     }
 }

src/DotNetCore.CAP.MySql/MySqlStorageTransaction.cs

@@ -46,7 +46,7 @@ public void EnqueueMessage(CapPublishedMessage message)
         {
             if (message == null) throw new ArgumentNullException(nameof(message));
 
-            var sql = $"INSERT INTO `{_prefix}.queue` values(@MessageId,@MessageType);";
+            var sql = $"INSERT INTO `{_prefix}.queue`(`MessageId`,`MessageType`) values(@MessageId,@MessageType);";
             _dbConnection.Execute(sql, new CapQueue {MessageId = message.Id, MessageType = MessageType.Publish},
                 _dbTransaction);
         }
@@ -55,7 +55,7 @@ public void EnqueueMessage(CapReceivedMessage message)
         {
             if (message == null) throw new ArgumentNullException(nameof(message));
 
-            var sql = $"INSERT INTO `{_prefix}.queue` values(@MessageId,@MessageType);";
+            var sql = $"INSERT INTO `{_prefix}.queue`(`MessageId`,`MessageType`) values(@MessageId,@MessageType);";
             _dbConnection.Execute(sql, new CapQueue {MessageId = message.Id, MessageType = MessageType.Subscribe},
                 _dbTransaction);
         }