python - Python-varchar的Postgresql无效字节序列-奇怪的行为

我已经编写了一个使用python线程函数的脚本。我认为问题与线程有关，因为当我从工作线程外部运行查询时，它工作正常。我试图在数据库中插入一些内容，但遇到了一些非常可怕的行为。
让我简化一下：
运行此操作：

cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
data = ("solaris-cdc", resultHOST[0], "UX10", 1,)
sql.execute(cmd, data)

运行此命令不起作用：

cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
data = ("solaris-cdc", resultHOST[0], "sdsdsdsdsdsd", 1,)
sql.execute(cmd, data)

以下是字段类型：
device=可变字符
host=可变字符
ux=可变字符
units=整数
这是我收到的错误：

Exception in thread Thread-1:
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/usr/local/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/cstanley/scripts/vip/sun_audit.py", line 37, in workon
    sql.execute(cmd, data)
DataError: invalid byte sequence for encoding "UTF8": 0x86
HINT:  This error can also happen if the byte sequence does not match the encoding expected by the server, which is controlled by "client_encoding".

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/threading.py", line 801, in __bootstrap_inner
    self.run()
  File "/usr/local/lib/python2.7/threading.py", line 754, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/cstanley/scripts/vip/sun_audit.py", line 37, in workon
    sql.execute(cmd, data)
InternalError: current transaction is aborted, commands ignored until end of transaction block

下面是完整的代码：

#!/usr/local/bin/python2.7
import sys, os, string, threading
import paramiko
import psycopg2
import time

#paramiko.util.log_to_file("sun_audit.log")

getCPU = "/usr/sbin/psrinfo -p"
getMEM = "/usr/sbin/prtconf | grep \"Memory\" | awk '{ print $3 }'"
getHOST = "hostname"

class bcolors:
    MAGENTA = '\033[95m'
    YELLOW = '\033[93m'
    ENDC = '\033[0m'

def workon(host,sql):

    #Connect to each host
    ssh = paramiko.SSHClient()
    key = paramiko.RSAKey.from_private_key_file("/home/cstanley/scripts/vip/cstanley")
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host, username='cstanley', pkey=key)

    #Run Commands
    stdinHOST, stdoutHOST, stderrHOST = ssh.exec_command(getHOST)
    stdinCPU, stdoutCPU, stderrCPU = ssh.exec_command(getCPU)
    stdinMEM, stdoutMEM, stderrMEM = ssh.exec_command(getMEM)

    with threading.Lock():

        resultHOST = stdoutHOST.readlines()
        #print "{0} {0} UX10 1".format(resultHOST[0].rstrip())
        cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
        data = ("solaris-cdc", resultHOST[0], "sdsdsdsdsdsd", 1,)
        sql.execute(cmd, data)

        resultCPU = stdoutCPU.readlines()
        ux40 = (int(resultCPU[0].rstrip()) - 1)
        if ux40 != 0:
            #print "{0} {0} UX40 {1}".format(resultHOST[0].rstrip(),ux40)
            cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
            data = ("solaris-cdc", resultHOST[0], "UX40", ux40,)
            sql.execute(cmd, data)

        resultMEM = stdoutMEM.readlines()
        ux30 = (int(resultMEM[0].rstrip()) / 1024 - 2) / 2
        #print "{0} {0} UX30 {1}".format(resultHOST[0].rstrip(),ux30)
        cmd = "INSERT INTO cstanley_temp (device, host, ux, units) VALUES (%s, %s, %s, %s);"
        data = ("solaris-cdc", resultHOST[0], "UX30", ux30,)
        sql.execute(cmd, data)

        ssh.close()

def main():

    #date = (time.strftime("%Y-%m-%d"))

    #Define our connection string
    conn_string = "host='REMOVED' dbname='REMOVED' user='REMOVED' password='REMOVED' connect_timeout=3"

    # print the connection string we will use to connect
    #print bcolors.MAGENTA + 'Connecting to database\n    ->%s' % (conn_string) + bcolors.ENDC + "\n"

    # get a connection, if a connect cannot be made an exception will be raised here
    conn = psycopg2.connect(conn_string)

    # conn.cursor will return a cursor object, you can use this cursor to perform queries
    sql = conn.cursor()
    print bcolors.YELLOW + "Inserting Solaris information into table.\n" + bcolors.ENDC

    with open('/home/cstanley/scripts/vip/sun_ip') as ip:
        hosts = ip.read().splitlines()

    threads = []
    for h in hosts:
        t = threading.Thread(target=workon, args=(h,sql,))
        t.start()
        threads.append(t)
    for t in threads:
        t.join()

    conn.commit()
    sql.close()
    conn.close()

if __name__ == "__main__":
    main()

想知道这里发生了什么。为什么当我输入UX10而不是输入sdsdsdsdsdsd时它会工作？我甚至尝试用solaris-cdc替换它，就像在查询的第一部分一样，但这也失败了。到底发生了什么事！?

最佳答案

错误似乎是这样说的：当PostgreSQL期望给出正确的UTF-8编码的unicode时，您正试图将一些Python 2strs和binary插入到varchar列中。它可能间歇性地来自resultHOST[0]字符串，或者您的代码中有一些不可见的控制字符在"sdsdsdsdsdsd"字符串中：

>>> print u"here be ma\x86gic"
here be magic

然而，真正的原因可能是根据psycopg2 documentation，
游标不是线程安全的：多线程应用程序可以从同一个连接创建多个游标，并且应该使用单个线程中的每个游标。详见Thread and process safety。
因此，您应该在每个线程上创建一个新的游标。不用将sql作为参数传递给workon，只需在workon方法中使用语句创建一个新的游标

sql = conn.cursor()

另外，不需要锁定，因为使用一个连接和多个游标是线程安全的。