python - 为什么 python Cassandra 驱动程序在声明为类字段时会失败？

Question

我目前正在使用一个服务器进程，该进程旨在连接到 Cassandra 数据库并向其转发信息。此过程被编写为一个类，我的目标是为该类创建一个可用于发送信息的 Cassandra session 。然而，我遇到了一个问题；当我在类 init 方法中创建 Cassandra session ，然后尝试在另一个方法中使用该 session 时，出现以下错误: errors={}, last_host=<server IP address> 。目前，我可以通过在每次调用该方法时创建一个新的 Cassandra session 来解决此问题，但这显然不是解决此问题的好方法。那么，我怎样才能创建一个可以在整个类里面一致使用的 Cassandra session 呢？

此代码不起作用:

from cassandra.cluster import Cluster
from multiprocessing import Process
class DataProcess(Process):

    def __init__(self):

        super(DataProcess,self).__init__()

        # Do a few other irrelevant things ...

        # Set up the Cassandra connection
        self.cluster = Cluster(contact_points=[CASSANDRA_IP])
        self.session = self.cluster.connect('some keyspace')
        print "Connected to cassandra."

    def callback(self,ch,method,props,body):
        prepared_statement = self.session.prepare("Some CQL statement...")
        bound_statement = prepared_statement.bind(some values)
        self.session.execute(bound_statement)

Output:
"Connected to cassandra."
errors={}, last_host=<server IP address>

这段代码确实有效，但这是一种愚蠢的方法:

from cassandra.cluster import Cluster
from multiprocessing import Process
class DataProcess(Process):

    def __init__(self):

        super(DataProcess,self).__init__()

        # Do a few irrelevant things ...


    def callback(self,ch,method,props,body):
        # Set up the Cassandra connection
        cluster = Cluster(contact_points=[CASSANDRA_IP])
        session = cluster.connect('some keyspace')
        prepared_statement = session.prepare("Some CQL statement...")
        bound_statement = prepared_statement.bind(some values)
        session.execute(bound_statement)

其他相关信息:

使用 python cassandra-driver 版本 2.5.1

Cassandra 数据库版本 2.1.8

编辑:答案 以下代码解决了该问题:

from cassandra.cluster import Cluster
from multiprocessing import Process
class DataProcess(Process):

    def __init__(self):

        super(DataProcess,self).__init__()
        self.cluster = None
        self.session = None
        # Do a few irrelevant things ...


    def callback(self,ch,method,props,body):
        # Set up the Cassandra connection
        cluster = Cluster(contact_points=[CASSANDRA_IP])
        session = cluster.connect('some keyspace')
        prepared_statement = session.prepare("Some CQL statement...")
        bound_statement = prepared_statement.bind(some values)
        session.execute(bound_statement)

    def run(self):
        self.cluster = Cluster(contact_points=[CASSANDRA_IP])
        self.session = self.cluster.connect('some keyspace')

Answer 1

您是否在 fork 之前创建集群和 session ？这可能会导致像您所看到的那样的问题。这里有一篇关于如何使用 python 驱动程序使用池分配工作的惊人文章 here 。这可能正是您正在寻找的。

如果不是，请留下更多关于如何运行流程的背景信息，因为在不知道流程生命周期的情况下很难重现。