L'accès à une connexion MySQL piscine de Python de multitraitement

Je suis en train de configurer une connexion MySQL piscine et de mon processus de travail d'accès déjà en place piscine au lieu de la configuration d'une nouvelle connexion à chaque fois.

Je suis confus si je passe le curseur base de données pour chaque processus, ou s'il y a une autre façon de faire? Ne devrait pas MySql.connecteur de faire la mise en commun automatiquement? Quand j'ai vérifier mes fichiers journaux, beaucoup, beaucoup de connexions sont ouvertes et fermées ... un pour chaque processus.

Mon code ressemble à ceci:

PATH = "/tmp"

class DB(object):
  def __init__(self):
    connected = False
    while not connected:
      try:
        cnxpool = mysql.connector.pooling.MySQLConnectionPool(pool_name = "pool1",
                                                          **config.dbconfig)
        self.__cnx = cnxpool.get_connection()
      except mysql.connector.errors.PoolError:
        print("Sleeping.. (Pool Error)")
        sleep(5)
      except mysql.connector.errors.DatabaseError:
        print("Sleeping.. (Database Error)")
        sleep(5)

    self.__cur = self.__cnx.cursor(cursor_class=MySQLCursorDict)

  def execute(self, query):
    return self.__cur.execute(query)

def isValidFile(self, name):
  return True

def readfile(self, fname):
  d = DB()
  d.execute("""INSERT INTO users (first_name) VALUES ('michael')""")

def main():
  queue = multiprocessing.Queue()
  pool = multiprocessing.Pool(None, init, [queue])
  for dirpath, dirnames, filenames in os.walk(PATH):

    full_path_fnames = map(lambda fn: os.path.join(dirpath, fn),
                           filenames)
    full_path_fnames = filter(is_valid_file, full_path_fnames)
    pool.map(readFile, full_path_fnames)

if __name__ == '__main__':
  sys.exit(main())

Peut-être que eric.lubow.org/2009/python/... est utile pour vous.

OriginalL'auteur ensnare | 2014-06-23

8

Tout d'abord, vous êtes en train de créer un autre pool de connexion pour chaque instance de votre DB classe. Les piscines ayant le même nom ne fait pas d'eux de la même piscine

De la la documentation:

Il n'est pas une erreur pour plusieurs piscines d'avoir le même nom. Une application qui doit distinguer les piscines par leur pool_name propriété doit créer chaque piscine avec un nom distinct.

En outre, le partage d'une connexion de base de données (ou pool de connexions) entre les différents processus serait une mauvaise idée (et j'en doute fortement, il serait même fonctionner correctement), de sorte que chaque processus en utilisant ses propres connexions est en fait ce que vous devriez viser pour.

Vous pourriez initialiser la piscine dans votre init initialiseur comme une variable globale et de l'utiliser à la place.

Exemple très simple:
```
from multiprocessing import Pool
from mysql.connector.pooling import MySQLConnectionPool
from mysql.connector import connect
import os

pool = None

def init():
    global pool
    print("PID %d: initializing pool..." % os.getpid())
    pool = MySQLConnectionPool(...)

def do_work(q):
    con = pool.get_connection()
    print("PID %d: using connection %s" % (os.getpid(), con))
    c = con.cursor()
    c.execute(q)
    res = c.fetchall()
    con.close()
    return res

def main():
    p = Pool(initializer=init)
    for res in p.map(do_work, ['select * from test']*8):
        print(res)
    p.close()
    p.join()

if __name__ == '__main__':
    main()
```
Ou tout simplement utiliser une connexion simple au lieu d'un pool de connexions, comme une seule connexion sera active dans chaque processus à un moment de toute façon.

Le nombre de utilisé simultanément des connexions est implicitement limité par la taille de la multiprocessing.Pool.

Ce n'est pas un bon moyen de le faire. Init sera appelée à partir de chaque travailleur séparément, fondamentalement, la génération d'une nouvelle mysql piscine pour chaque travailleur chaque fois.
Il serait de créer connexion de la piscine pour chaque travailleur proccess lorsque le processus de travail est commencé. Une connexion par travailleur peut être plus approprié puis d'une piscine par travailleur selon le cas d'utilisation, comme je l'ai explaind en bas de ma réponse. Mais la question est explicite sur l'utilisation d'un pool de connexion avec le multitraitement, c'est pourquoi j'ai gardé la piscine dans mon exemple.
La bonne méthode serait d'utiliser un Pool de chaque processus peut accéder à ce qui est ce que l'OP a demandé. Faire une piscine séparée pour chaque processus est redondant et ouvre la voie à de trop nombreuses connexions qui est exactement le problème de l'OP est de mentionner.
La seule façon d'utiliser un seul à la piscine, avec de nombreux processus est d'avoir un processus spécifique qui fait tout le db d'accès de communiquer avec elle à l'aide d'une file d'attente, mais qui impliquerait une charge importante pour le décapage et la unpickling les données. Comme je l'avais dit, j'avais juste un couple de travailleurs avec une connexion de chaque, qui est la même dans nubmer de connexions à l'aide d'une piscine avec le même nombre de connexions. Mais si vous pensez à une solution avec une piscine serait mieux, vous devriez vous suffit d'envoyer votre propre réponse.
Malheureusement, je ne suis pas sûr de comment résoudre ce problème d'élégance. Je suis d'accord que peut-être la solution la plus pragmatique est d'ouvrir et de fermer une nouvelle connexion à chaque fois qu'un travailleur est appelé. Je voulais juste faire remarquer que mysql piscine pour chaque travailleur n'est pas une solution à ce problème.

OriginalL'auteur mata

#!/usr/bin/python
# -*- coding: utf-8 -*-
import time
import mysql.connector.pooling
dbconfig = {
"host":"127.0.0.1",
"port":"3306",
"user":"root",
"password":"123456",
"database":"test",
}
class MySQLPool(object):
"""
create a pool when connect mysql, which will decrease the time spent in 
request connection, create connection and close connection.
"""
def __init__(self, host="172.0.0.1", port="3306", user="root",
password="123456", database="test", pool_name="mypool",
pool_size=3):
res = {}
self._host = host
self._port = port
self._user = user
self._password = password
self._database = database
res["host"] = self._host
res["port"] = self._port
res["user"] = self._user
res["password"] = self._password
res["database"] = self._database
self.dbconfig = res
self.pool = self.create_pool(pool_name=pool_name, pool_size=pool_size)
def create_pool(self, pool_name="mypool", pool_size=3):
"""
Create a connection pool, after created, the request of connecting 
MySQL could get a connection from this pool instead of request to 
create a connection.
:param pool_name: the name of pool, default is "mypool"
:param pool_size: the size of pool, default is 3
:return: connection pool
"""
pool = mysql.connector.pooling.MySQLConnectionPool(
pool_name=pool_name,
pool_size=pool_size,
pool_reset_session=True,
**self.dbconfig)
return pool
def close(self, conn, cursor):
"""
A method used to close connection of mysql.
:param conn: 
:param cursor: 
:return: 
"""
cursor.close()
conn.close()
def execute(self, sql, args=None, commit=False):
"""
Execute a sql, it could be with args and with out args. The usage is 
similar with execute() function in module pymysql.
:param sql: sql clause
:param args: args need by sql clause
:param commit: whether to commit
:return: if commit, return None, else, return result
"""
# get connection form connection pool instead of create one.
conn = self.pool.get_connection()
cursor = conn.cursor()
if args:
cursor.execute(sql, args)
else:
cursor.execute(sql)
if commit is True:
conn.commit()
self.close(conn, cursor)
return None
else:
res = cursor.fetchall()
self.close(conn, cursor)
return res
def executemany(self, sql, args, commit=False):
"""
Execute with many args. Similar with executemany() function in pymysql.
args should be a sequence.
:param sql: sql clause
:param args: args
:param commit: commit or not.
:return: if commit, return None, else, return result
"""
# get connection form connection pool instead of create one.
conn = self.pool.get_connection()
cursor = conn.cursor()
cursor.executemany(sql, args)
if commit is True:
conn.commit()
self.close(conn, cursor)
return None
else:
res = cursor.fetchall()
self.close(conn, cursor)
return res
if __name__ == "__main__":
mysql_pool = MySQLPool(**dbconfig)
sql = "select * from store WHERE create_time < '2017-06-02'"
p = Pool()
for i in range(5):
p.apply_async(mysql_pool.execute, args=(sql,))

Code ci-dessus crée un pool de connexions au début, et d'obtenir des connexions dans execute(), une fois le pool de connexion a été créé, le travail est de rester, car la piscine est créé qu'une seule fois, il vous permettra d'économiser le temps de faire la demande pour une connexion à chaque fois que vous voulez vous connecter à MySQL.
Espérons que cela aide!

OriginalL'auteur buxizhizhoum

0

Vous avez créé plusieurs DB d'instance d'objet. Dans mysql.connector.pooling.py, pool_name n'est qu'un attribut pour vous permettre de faire qui de la piscine. Il n'existe aucune correspondance dans la base de données mysql piscine.

Donc, vous devez créer plusieurs instance DB en def readfile(), alors vous aurez plusieurs pool de connexion.

Un Singleton est utile dans ce cas.

(J'ai passé plusieurs heures à le retrouver. Dans la Tornade cadre, chaque http get créer un nouveau gestionnaire, ce qui conduit à la fabrication d'une nouvelle connexion.)

OriginalL'auteur ThomasYe

Vous devez vous connecter pour publier un commentaire.