从 SQLite 接收 .NET 中的数据库更新事件

Question

我最近发现了 SQLite 的强大之处，特别是 SQLite 的 .NET 包装器，位于 http://sqlite.phxsoftware.com /。

现在，假设我正在开发将在同一网络上的多台计算机上运行的软件。没什么疯狂的，可能只有 5 或 6 台机器。每个软件实例都将访问存储在共享目录中的文件中的 SQLite 数据库（这是一个坏主意吗？如果是这样，请告诉我！）。

如果一个实例更新数据库文件，是否有一种方法可以通知应用程序的每个实例？一种明显的方法是使用 FileSystemWatcher 类，将整个数据库读入 DataSet，然后...你知道...枚举整个内容以查看新内容...但是是的，实际上这看起来相当愚蠢。有 SQLite 更新提供者这样的东西吗？

这作为一个问题有意义吗？当谈到 ADO.NET 时，我也几乎是一个新手，所以我可能从完全错误的角度来处理这个问题。

Answer 1

通过网络使用 SQLite 并不是一个好主意。 此处查看 SQLite 自己的建议。

客户端-服务器数据库将更加可靠，并且还可以解决您的通知问题。例如，PostgreSQL 通过 NOTIFY 和 LISTEN 语句具有客户端间信号机制，可以直接从客户端或从函数、存储过程或触发器内部使用。

即使您决定使用 SQLite，也不要使用文件监视 API。由于文件系统内部深处的竞争条件，它们在 Windows 上完全被破坏。来自 FileSystemWatcher 的 MSDN 条目：

请注意，由于与 Windows 操作系统的依赖性，当错过事件或超出缓冲区大小时，FileSystemWatcher 不会引发错误事件。

它提供了缓解这种情况的建议，但没有一个提供任何可靠性保证。

Answer 2

虽然没有任何内置机制可以挂钩某种更改事件，但您可以跨多个进程检测 SQLite 数据库中的更改。

SQLite 数据库不应通过网络共享进行共享，原因有多种，创建者在此进行了解释：https:// /www.sqlite.org/useovernet.html

如果您有一台运行 TCP 客户端服务器应用程序的服务器来接收您的请求并发送数据作为响应，那么它会更加高效和一致。

轮询文件头

SQLite 文件头包含字段 文件更改计数器，该字段每当数据库或其表已被修改。我们可以在标头上使用轮询来检测更改。我不知道为什么这么多人反对轮询，这在低级编程中是完全正常的，如果软件或硬件不提供任何中断，无论如何你都会陷入轮询。

SQLiteHeader struct

/// <summary>
/// The first 100 bytes of the database file comprise 
/// the database file header. The database file header 
/// is divided into fields as shown by the <see cref="SQLiteHeader"/> struct. 
/// All multibyte fields in the database file header 
/// are stored with the most significant byte first (big-endian).
/// The <see cref="SQLiteHeader"/> already checks for endianess and
/// converts the mulitbyte fields for you.
/// </summary>
[StructLayout(LayoutKind.Explicit)]
unsafe public struct SQLiteHeader
{       
    public static SQLiteHeader FromFile(string fileName, out SQLiteHeaderError error)
    {
        return SQLiteHeaderParser.FromFile(fileName, out error);
    }


    /// <summary>
    /// The header string: "SQLite format 3\000"
    /// </summary>
    [FieldOffset(0)]
    [MarshalAs(UnmanagedType.ByValTStr, SizeConst = 16)]
    public string HeaderString;
    /// <summary>
    /// The database page size in bytes. 
    /// Must be a power of two between 512 and 32768 inclusive, 
    /// or the value 1 representing a page size of 65536.
    /// </summary>
    [FieldOffset(16)]
    public short PageSize;
    /// <summary>
    /// File format write version. 1 for legacy; 2 for WAL.
    /// </summary>
    [FieldOffset(18)]
    public byte WriteVersion;
    /// <summary>
    /// File format read version. 1 for legacy; 2 for WAL.
    /// </summary>
    [FieldOffset(19)]
    public byte ReadVersion;
    /// <summary>
    /// Bytes of unused "reserved" space at the end of each page. Usually 0.
    /// </summary>
    [FieldOffset(20)]
    public byte Reserved;
    /// <summary>
    /// Maximum embedded payload fraction. Must be 64.
    /// </summary>
    [FieldOffset(21)]
    public byte MaxPayloadFract;
    /// <summary>
    /// Minimum embedded payload fraction. Must be 32.
    /// </summary>
    [FieldOffset(22)]
    public byte MinPayloadFract;
    /// <summary>
    /// Leaf payload fraction. Must be 32.
    /// </summary>
    [FieldOffset(23)]
    public byte LeafPayloadFract;
    /// <summary>
    /// File change counter.
    /// </summary>
    [FieldOffset(24)]
    public Int32 FileChangeCounter;
    /// <summary>
    /// Size of the database file in pages. 
    /// The "in-header database size".
    /// </summary>
    [FieldOffset(28)]
    public Int32 InHeaderDatabaseSize;
    /// <summary>
    /// Page number of the first freelist trunk page.
    /// </summary>
    [FieldOffset(32)]
    public Int32 FirstFreeListTrunkPage;
    /// <summary>
    /// Total number of freelist pages.
    /// </summary>
    [FieldOffset(36)]
    public Int32 NumFreeListPages;
    /// <summary>
    /// The schema cookie.
    /// </summary>
    [FieldOffset(40)]
    public Int32 SchemaCookie;
    /// <summary>
    /// The schema format number. 
    /// Supported schema formats are 1, 2, 3, and 4.
    /// </summary>
    [FieldOffset(44)]
    public Int32 SchemaFormatNumber;
    /// <summary>
    /// Default page cache size.
    /// </summary>
    [FieldOffset(48)]
    public Int32 PageCacheSize;
    /// <summary>
    /// The page number of the largest root b-tree 
    /// page when in auto-vacuum or incremental-vacuum modes, 
    /// or zero otherwise.
    /// </summary>
    [FieldOffset(52)]
    public Int32 MaxRootBTreePage;
    /// <summary>
    /// The database text encoding. 
    /// A value of 1 means UTF-8. 
    /// A value of 2 means UTF-16le. 
    /// A value of 3 means UTF-16be.
    /// </summary>
    [FieldOffset(56)]
    public Int32 TextEncoding;
    /// <summary>
    /// The "user version" as read and set 
    /// by the user_version pragma.
    /// </summary>
    [FieldOffset(60)]
    public Int32 UserVersion;
    /// <summary>
    /// True (non-zero) for incremental-vacuum mode. 
    /// False (zero) otherwise.
    /// </summary>
    [FieldOffset(64)]
    public Int32 IncrementalVacuumMode;
    /// <summary>
    /// The "Application ID" set by 
    /// PRAGMA application_id.
    /// </summary>
    [FieldOffset(68)]
    public Int32 ApplicationId;
    /// <summary>
    /// Reserved for expansion. Must be zero.
    /// </summary>
    [FieldOffset(72)]
    [MarshalAs(UnmanagedType.ByValTStr, SizeConst = 20)]
    public string ReservedExpansion;
    /// <summary>
    /// The version-valid-for number.
    /// </summary>
    [FieldOffset(92)]
    public Int32 VersionValidFor;
    /// <summary>
    /// SQLITE_VERSION_NUMBER
    /// </summary>
    [FieldOffset(96)]
    public Int32 SQLiteVersionNumber;        
}

SQLiteHeaderParser 类

internal static class SQLiteHeaderParser
{
    internal static readonly byte[] MagicHeaderBytes = new byte[] {
        0x53, 0x51, 0x4c, 0x69,
        0x74, 0x65, 0x20, 0x66,
        0x6f, 0x72, 0x6d, 0x61,
        0x74, 0x20, 0x33, 0x00
    };


    private static void ToLittleEndian(byte[] buffer, int offset, int length)
    {
        switch (length)
        {
            case 2:
                {
                    ushort num = BitConverter.ToUInt16(buffer, offset);
                    num = (ushort)(
                        ((num & 0xff) >> 8) |
                         (num << 8)
                    );
                    var bytes = BitConverter.GetBytes(num);
                    bytes.CopyTo(buffer, offset);
                    break;
                }
            case 4:
                {
                    uint num = BitConverter.ToUInt32(buffer, offset);
                    num = (uint)(
                        ((num & 0xff000000) >> 24) |
                        ((num & 0x00ff0000) >> 8) |
                        ((num & 0x0000ff00) << 8) |
                         (num << 24)
                    );
                    var bytes = BitConverter.GetBytes(num);
                    bytes.CopyTo(buffer, offset);
                    break;
                }
        }
    }


    private static int GetOffset(string name)
    {
        return Marshal.OffsetOf<SQLiteHeader>(name).ToInt32();
    }        


    private static void SetEndianess(byte[] buffer)
    {
        if (!(BitConverter.IsLittleEndian)) {
            return;
        }

        SQLiteHeader h = new SQLiteHeader();

        ToLittleEndian(buffer, GetOffset(nameof(h.PageSize)),               Marshal.SizeOf(h.PageSize));
        ToLittleEndian(buffer, GetOffset(nameof(h.FileChangeCounter)),      Marshal.SizeOf(h.FileChangeCounter));
        ToLittleEndian(buffer, GetOffset(nameof(h.InHeaderDatabaseSize)),   Marshal.SizeOf(h.InHeaderDatabaseSize));
        ToLittleEndian(buffer, GetOffset(nameof(h.FirstFreeListTrunkPage)), Marshal.SizeOf(h.FirstFreeListTrunkPage));
        ToLittleEndian(buffer, GetOffset(nameof(h.NumFreeListPages)),       Marshal.SizeOf(h.NumFreeListPages));
        ToLittleEndian(buffer, GetOffset(nameof(h.SchemaCookie)),           Marshal.SizeOf(h.SchemaCookie));
        ToLittleEndian(buffer, GetOffset(nameof(h.SchemaFormatNumber)),     Marshal.SizeOf(h.SchemaFormatNumber));
        ToLittleEndian(buffer, GetOffset(nameof(h.PageCacheSize)),          Marshal.SizeOf(h.PageCacheSize));
        ToLittleEndian(buffer, GetOffset(nameof(h.MaxRootBTreePage)),       Marshal.SizeOf(h.MaxRootBTreePage));
        ToLittleEndian(buffer, GetOffset(nameof(h.TextEncoding)),           Marshal.SizeOf(h.TextEncoding));
        ToLittleEndian(buffer, GetOffset(nameof(h.PageCacheSize)),          Marshal.SizeOf(h.PageCacheSize));
        ToLittleEndian(buffer, GetOffset(nameof(h.UserVersion)),            Marshal.SizeOf(h.UserVersion));
        ToLittleEndian(buffer, GetOffset(nameof(h.IncrementalVacuumMode)),  Marshal.SizeOf(h.IncrementalVacuumMode));
        ToLittleEndian(buffer, GetOffset(nameof(h.ApplicationId)),          Marshal.SizeOf(h.ApplicationId));
        ToLittleEndian(buffer, GetOffset(nameof(h.VersionValidFor)),        Marshal.SizeOf(h.VersionValidFor));
        ToLittleEndian(buffer, GetOffset(nameof(h.SQLiteVersionNumber)),    Marshal.SizeOf(h.SQLiteVersionNumber));
    }


    private static byte[] PrepareHeader(BinaryReader reader, out SQLiteHeaderError error)
    {
        int hdrSize = Marshal.SizeOf<SQLiteHeader>();
        byte[] buffer = reader.ReadBytes(hdrSize);

        var loadedMagicHeader = buffer.Take(MagicHeaderBytes.Length).ToArray();
        if (!(loadedMagicHeader.SequenceEqual(MagicHeaderBytes))) {
            System.Diagnostics.Debug.WriteLine("magic number miss match -> invalid header");
            error = SQLiteHeaderError.InvalidMagicNumber;
            return new byte[0];
        }

        SetEndianess(buffer);
        error = SQLiteHeaderError.Success;
        return buffer;
    }


    private static SQLiteHeader BufferToHeader(byte[] buffer)
    {
        GCHandle gchHeader = GCHandle.Alloc(buffer, GCHandleType.Pinned);
        SQLiteHeader hdr = Marshal.PtrToStructure<SQLiteHeader>(gchHeader.AddrOfPinnedObject());
        gchHeader.Free();

        return hdr;
    }


    public static SQLiteHeader FromFile(string filePath, out SQLiteHeaderError error)
    {
        if (!(File.Exists(filePath))) {
            System.Diagnostics.Debug.WriteLine("invalid file path");
            error = SQLiteHeaderError.FileNotFound;
            return default(SQLiteHeader);
        }

        int hdrSize = Marshal.SizeOf<SQLiteHeader>();

        var fInfo = new FileInfo(filePath);
        if (fInfo.Length < hdrSize) {
            System.Diagnostics.Debug.WriteLine("not enough bytes");
            error = SQLiteHeaderError.InvalidHeaderSize;
            return default(SQLiteHeader);
        }

        byte[] buffer;
        try
        {
            using (var stream = new FileStream(
                filePath, 
                FileMode.Open, 
                FileAccess.Read, 
                FileShare.ReadWrite, hdrSize))
            {
                using (var reader = new BinaryReader(stream))
                {
                    buffer = PrepareHeader(reader, out error);
                    if (buffer.Length < 1) {                   
                        return default(SQLiteHeader);
                    }
                }
            }                
        }
        catch (Exception ex)
        {
            System.Diagnostics.Debug.WriteLine(ex.Message);
            error = SQLiteHeaderError.Undefined;
            return default(SQLiteHeader);
        }

        SQLiteHeader header = BufferToHeader(buffer);
        return header;
    }
}

SQLiteHeaderError 枚举

[Flags]
public enum SQLiteHeaderError
{
    Success             = 0,
    FileNotFound        = 1 << 1,
    InvalidHeaderSize   = 1 << 2,
    InvalidMagicNumber  = 1 << 3,
    Undefined           = 1 << 8
}

SQLiteChangeMonitor 类

public class SQLiteChangeMonitor : IDisposable
{
    private string          _sqliteFile;
    private SQLiteHeader    _sqliteHeader;

    private Timer _pollingTimer;

    private int _fileChangeCounter = -1;

    private static CancellationTokenSource  _cancelTokenSource;
    private static CancellationToken        _cancelToken;       


    private event EventHandler _onChangeDetected;
    public event EventHandler ChangeDetected
    {
        add {
            _onChangeDetected += value;
        }
        remove { 
            _onChangeDetected -= value; 
        }
    }


    //
    // Check for any changes in the header and restart timer.
    //
    private void TimerCallback(object state)
    {
        AutoResetEvent autoEvent = (AutoResetEvent)state;
        if (autoEvent == null) {
            return;
        }
        CheckForChanges();
        autoEvent.Set();
    }


    //
    // Check if the file change counter of the header has changed
    // and fire event to signal change.
    //
    private void CheckForChanges()
    {
        if (_cancelToken.IsCancellationRequested) {
            return;
        }

        try
        {
            SQLiteHeaderError err;
            _sqliteHeader = SQLiteHeader.FromFile(_sqliteFile, out err);
            int fcc = _sqliteHeader.FileChangeCounter;
            if (fcc != _fileChangeCounter)
            {
                _fileChangeCounter = fcc;
                _onChangeDetected?.Invoke(this, EventArgs.Empty);
            }
        }
        catch (Exception ex) 
        {
            // Implement your error handling here.
            System.Diagnostics.Debug.WriteLine(ex.Message);
        }
    }   


    //
    // Start the System.Threading.Timer
    //
    private void StartPolling(int timeout)
    {
        AutoResetEvent autoResetEvent = new AutoResetEvent(false);

        _pollingTimer = new Timer(TimerCallback, autoResetEvent, 0, timeout);
        autoResetEvent.WaitOne();
    }


    //
    // Get current change counter from file header to 
    // avoid missfire when starting the polling.
    //
    private void SetInitalChangeCounter()
    {
        SQLiteHeaderError err;
        _sqliteHeader = SQLiteHeader.FromFile(_sqliteFile, out err);
        _fileChangeCounter = _sqliteHeader.FileChangeCounter;
    }


    /// <summary>
    /// Starts polling for database changes.
    /// </summary>
    /// <param name="pollingInterval">
    /// The polling interval in milliseconds</param>
    public void Start(int pollingInterval = 1000)
    {
        SetInitalChangeCounter();
        StartPolling(pollingInterval);
    }


    /// <summary>
    /// Stops polling for changes.
    /// </summary>
    public void Stop()
    {
        _cancelTokenSource?.Cancel();
    }
    
    
    //
    // Set up cancellation token to end polling.
    //
    private void InitToken()
    {
        _cancelTokenSource = new CancellationTokenSource();
        _cancelToken = _cancelTokenSource.Token;
    }


    public SQLiteChangeMonitor(string databaseFile)
    {
        if (string.IsNullOrEmpty(databaseFile)) {
            throw new ArgumentNullException(
                nameof(databaseFile), 
                "Database file path cannot be null or empty."
            );
        }

        if (!(File.Exists(databaseFile))) {
            throw new FileNotFoundException(
                "Unable to locate file.", 
                databaseFile
            );
        }

        _sqliteFile = databaseFile;

        InitToken();
    }


    ~SQLiteChangeMonitor() { Dispose(false); }
    public void Dispose() { Dispose(true); }
    private void Dispose(bool disposing)
    {
        if (disposing)
        {
            GC.SuppressFinalize(this);
        }
        
        _cancelTokenSource?.Cancel();            
        _pollingTimer?.Dispose();
    }
}

---

高频轮询

SQLiteChangeMonitor 类使用默认轮询间隔为 1000 毫秒 (1 秒)。对于高频轮询间隔，我建议使用此答案中描述的模式： https://stackoverflow.com/a/23341005/22191764缺点

---

得多

这不提供任何有关修改内容的信息，但轮询标头然后更新数据库视图然后始终对数据库运行查询并最终更新仍然要高效 你的观点。