使用 `ActiveRecord with_connection do` 和 ActionController::Live 时出现线程错误

Posted 2023-02-16

【中文标题】使用 `ActiveRecord with_connection do` 和 ActionController::Live 时出现线程错误

【英文标题】：Threading error when using `ActiveRecord with_connection do` & ActionController::Live

【发布时间】：2015-10-10 16:48:07

【问题描述】：

主要编辑：自从最初发现这个问题以来，我已将其缩减为以下内容。我认为现在这是对问题的略微更准确的描述。因此，对 OP 的评论可能并不完全相关。

编辑在 rails/puma 项目中发布的轻微修改版本：https://github.com/rails/rails/issues/21209、https://github.com/puma/puma/issues/758

编辑现在用 OS X 和 Rainbows 复制

总结： 在使用 Puma 并运行长时间运行的连接时，我一直收到与 ActiveRecord 跨线程连接相关的错误。这体现在像 message type 0x## arrived from server while idle 这样的消息和一个锁定（崩溃）的服务器。

设置：

Ubuntu 15 / OSX Yosemite PostgreSQL (9.4) / mysql (mysqld 5.6.25-0ubuntu0.15.04.1) Ruby - MRI 2.2.2p95 (2015-04-13 revision 50295) [x86_64-linux] / Rubinius rbx-2.5.8 导轨 (4.2.3, 4.2.1) 彪马 (2.12.2, 2.11) pg (pg-0.18.2) / mysql2

请注意，并非上述版本的所有组合都已尝试过。第一个列出的版本是我目前正在测试的版本。

rails new issue-test 添加路由get 'events' => 'streaming#events' 添加控制器streaming_controller.rb 设置数据库内容（pool: 2，但可以看到不同的池大小）

代码：

class StreamingController < ApplicationController

  include ActionController::Live

  def events
    begin
      response.headers["Content-Type"] = "text/event-stream"
      sse = SSE.new(response.stream)
      sse.write( :data => 'starting' , :event => :version_heartbeat)
      ActiveRecord::Base.connection_pool.release_connection
      while true do
        ActiveRecord::Base.connection_pool.with_connection do |conn|
          ActiveRecord::Base.connection.query_cache.clear
          logger.info 'START'
          conn.execute 'SELECT pg_sleep(3)'
          logger.info 'FINISH'
          sse.write( :data => 'continuing', :event => :version_heartbeat)
          sleep 0.5
         end
      end
    rescue IOError
    rescue ClientDisconnected
    ensure
      logger.info 'Ensuring event stream is closed'
      sse.close
    end
    render nothing: true
  end
end

彪马配置：

workers 1
threads 2, 2
#...
bind "tcp://0.0.0.0:9292"

#...
activate_control_app

on_worker_boot do
  require "active_record"
  ActiveRecord::Base.connection.disconnect! rescue ActiveRecord::ConnectionNotEstablished
  ActiveRecord::Base.establish_connection(YAML.load_file("#app_dir/config/database.yml")[rails_env])
end

运行服务器puma -e production -C path/to/puma/config/production.rb

测试脚本：

#!/bin/bash

timeout 30 curl -vS http://0.0.0.0/events &
timeout 5 curl -vS http://0.0.0.0/events &
timeout 30 curl -vS http://0.0.0.0/events

这合理一致地导致应用程序服务器的完全锁定（在 PostgreSQL 中，请参阅注释）。可怕的消息来自libpq：

message type 0x44 arrived from server while idle
message type 0x43 arrived from server while idle
message type 0x5a arrived from server while idle
message type 0x54 arrived from server while idle

在“现实世界”中，我有很多额外的元素，问题随机出现。我的研究表明，这条消息来自libpq，是“通信问题，可能使用不同线程中的连接”的潜台词。最后，在写这篇文章的时候，我在任何日志中都没有一条消息的情况下锁定了服务器。

所以，问题：

我遵循的模式在某些方面不合法吗？我错过了什么[sed|理解]？ 这里使用数据库连接的“标准”是什么，可以避免这些问题？ 您能找到一种可靠地重现这一点的方法吗？

或

这里的根本问题是什么？我该如何解决？

---

MySQL

如果运行 MySQL，消息有点不同，应用程序恢复（虽然我不确定它是否处于某种未定义状态）：

F, [2015-07-30T14:12:07.078215 #15606] FATAL -- : 
ActiveRecord::StatementInvalid (Mysql2::Error: This connection is in use by: #<Thread:0x007f563b2faa88@/home/dev/.rbenv/versions/2.2.2/lib/ruby/gems/2.2.0/gems/actionpack-4.2.3/lib/action_controller/metal/live.rb:269 sleep>: SELECT  `tasks`.* FROM `tasks`  ORDER BY `tasks`.`id` ASC LIMIT 1):

【问题讨论】：

来自 ActiveRecord 文档，使用 with_connection 依赖于作为参数完成传递的块。你确定它完成了吗？用 Base.connection 或 checkout 处理连接怎么样？

@Grasshopper - 谢谢！我担心这会在请求的整个生命周期（小时）内保持连接打开，从而很快吃掉我的连接池？我想它可能无法完成的方式是如果 sse.write 由于某种原因而阻塞并且线程只是坐在那里，即如果连接已经消失并且由于某种原因它没有返回？ （也就是说，我不确定这是否完全解释了来自 libpq 的基于线程问题的消息）。 （将朝那个方向尝试一些事情）

确实，您描述的问题可能发生在未释放连接的情况下。您可以尝试从 with_connection 块中删除对 sse.write 的调用吗？

另一个问题是：您真的需要启动连接吗，因为从 AR 文档看来 User.find 会为您完成。

你为什么不把所有这些东西都移到后台，然后把_user_id推到一个队列（可能是redis或rabbitmq）中，后台的东西会消耗掉，我不是确定你想根据网络请求打开和关闭连接，否则你会很快用完套接字

【参考方案1】：

警告：将“答案”解读为“似乎有所作为”

---

如果我将控制器块更改为如下所示，则不会出现问题：

begin
  #...
  while true do
    t = Thread.new do #<<<<<<<<<<<<<<<<<
        ActiveRecord::Base.connection_pool.with_connection do |conn|
            #...
        end
     end
     t.join #<<<<<<<<<<<<<<<<<
  end
  #...
rescue IOError
#...

但我不知道这是否真的解决了问题，或者只是让它变得极不可能。我也无法真正理解为什么这会有所作为。

将其发布为解决方案以防万一，但仍在挖掘问题。

【讨论】：

所以我也有一个长时间运行的过程，这也是我使用它的方式。我的问题是，为什么每次迭代都启动一个线程？启动线程并在循环外检查每个客户端的连接是否有意义？