【发布时间】：2015-01-28 01:34:12
【问题描述】：
我有一个应用程序管理的 JPA 设置（我第一次使用 JPA）。
我有一个读者线程和一个作者线程。
阅读器为每个查询或事务创建和关闭实体管理器。
编写器线程创建一个实体管理器并读取阻塞队列，以便它知道所有托管实体，因为我有很多一对多/父子关系，如果我有级联类型会导致重复异常（ com.mysql.jdbc.exceptions.jdbc4.MySQLIntegrityConstraintViolationException），或者如果我在每次实体写入时打开和关闭实体管理器，则没有级联导致瞬态实体异常（即实体管理器不知道父实体）。
我遇到的问题是，如果我通过 db writer 线程进行持久化，然后通过 sql 查询进行查询，则 db reader 线程返回 null，即使实体已持久保存在数据库 (mysql) 中并且两个实体管理器都是从相同的实体管理器工厂。如果我在每次插入后关闭数据库写入器线程中的实体管理器，则读取器线程确实知道插入的实体，但是写入器不知道先前插入的父实体并抛出 MySQLIntegrityConstraintViolationException。我相信关闭实体管理器似乎会分离所有实体并清除实体管理器的持久性上下文，但如果我通过@PostPersist 注释在方法中执行此操作，它似乎无助于在阅读器线程中创建任何实体管理器知道单个实体管理器在编写器线程中插入的实体。
是否有任何方法可以让在读取线程中创建的实体管理器知道写入线程所做的插入/更新？ （尝试驱逐所有，没有太多运气）。迁移到 EJB 和容器管理的 JPA 设置是一个很大的变化，我相信我还必须在服务器上运行 apache 来管理 EJB。
我尝试了一些事情，但运气不佳
- 在持久化时分离实体
- 持续驱逐实体
- 父实体的级联持久性
- 持续刷新实体
- 覆盖实体的等于和哈希码
- 读取时绕过缓存
实体管理器工厂
Map<String, String> properties = new HashMap<>();
properties.put("hibernate.hbm2ddl.auto", createMode);
properties.put("hibernate.connection.driver_class", ConfigUtil.combined().getString("db.driver"));
properties.put("hibernate.dialect", ConfigUtil.combined().getString("db.dialect"));
properties.put("hibernate.connection.url", ConfigUtil.combined().getString("db.url"));
properties.put("hibernate.connection.username", ConfigUtil.combined().getString("db.username"));
properties.put("hibernate.connection.password", ConfigUtil.combined().getString("db.password"));
properties.put("hibernate.ejb.naming_strategy", "org.hibernate.cfg.ImprovedNamingStrategy");
try {
PersistUtilHelper emh = new PersistUtilHelper(properties);
示例实体
public abstract class EntityBase implements Serializable {
@Id
@Column(columnDefinition = "BINARY(16)", length = 16, updatable = false, nullable = false)
public UUID getId() {
ensureId();
return id;
}
@Version
private Long version;
public Long getVersion() {
return version;
}
public void setVersion(Long version) {
this.version = version;
}
@Override
public boolean equals(Object o) {
// generated by IDEA
if (this == o)
return true;
if (!(o instanceof EntityBase))
return false;
EntityBase that = (EntityBase) o;
return id.equals(that.id);
}
@Override
public int hashCode() {
ensureId();
return id.hashCode();
}
// JPA
protected EntityBase() {
}
protected void setId(UUID id) {
this.id = id;
}
private void ensureId() {
if (id == null)
id = UUID.randomUUID();
}
private UUID id;
}
@Entity
@Table(name = "listing", uniqueConstraints = { @UniqueConstraint(columnNames = { "base", "quote", "prompt" }),
@UniqueConstraint(columnNames = { "base", "quote" }) })
public class Listing extends EntityBase {
@ManyToOne(optional = false)
//@Column(unique = true)
public Asset getBase() {
return base;
}
@PostPersist
private void postPersist() {
// PersistUtil.clear();
// PersistUtil.refresh(this);
//PersistUtil.merge(this);
// PersistUtil.close();
//PersistUtil.evict(this);
}
@ManyToOne(optional = false)
//@Column(unique = true)
public Asset getQuote() {
return quote;
}
@Nullable
//@Column(unique = true)
public String getPrompt() {
return prompt;
}
@Override
public boolean equals(Object obj) {
if (obj instanceof Listing) {
Listing listing = (Listing) obj;
if (!listing.getBase().equals(getBase())) {
return false;
}
if (!listing.getQuote().equals(getQuote())) {
return false;
}
return true;
}
return false;
}
@Override
public int hashCode() {
return getQuote().hashCode() + getBase().hashCode();
}
作家线程
public class DatabaseWriter implements Runnable {
private static final int defaultBatchSize = 20;
private static boolean running = false;
private static boolean shutdown = false;
private static ExecutorService service;
private static FutureTask persitanceTask = null;
private static BlockingQueue<EntityBase[]> blockingQueue;
public DatabaseWriter(BlockingQueue<EntityBase[]> blockingqueue) {
this.blockingQueue = blockingqueue;
}
@Override
public void run() {
EntityBase[] entities = null;
boolean persited = true;
while (!shutdown) {
try {
entities = blockingQueue.take();
for (EntityBase entity : entities)
PersistUtil.persist(entity);
}
catch (Exception e) {
} finally {
}
}
PersistUtilHelper.closeEntityManager();
}
}
样本阅读
Listing listing = PersistUtil.queryZeroOne(Listing.class, "select a from Listing a where base=?1 and quote=?2", base, quote);
读者线程
public static <T> T queryZeroOne(Class<T> resultType, String queryStr, Object... params) {
try {
//EntityManager em = createEntityManager();
final TypedQuery<T> query = PersistUtilHelper.createQuery(queryStr, resultType);
//query.setHint("org.hibernate.cacheable", false);
// query.setHint("javax.persistence.cache.retrieveMode", javax.persistence.CacheRetrieveMode.BYPASS);
//query.setHint("org.hibernate.cacheMode", CacheMode.IGNORE);
if (params != null) {
for (int i = 0; i < params.length; i++) {
Object param = params[i];
query.setParameter(i + 1, param); // JPA uses 1-based indexes
}
}
try {
return query.getSingleResult();
} catch (NoResultException x) {
return null;
}
} finally {
PersistUtilHelper.closeEntityManager();
}
}
辅助工具
public class PersistUtilHelper {
private static EntityManagerFactory emf = null;
private static final ThreadLocal<EntityManager> threadLocal = new ThreadLocal<EntityManager>();
private static ConcurrentHashMap<String, EntityManager> entityManagers = new ConcurrentHashMap<String, EntityManager>();
PersistUtilHelper(Map<String, String> properties) {
emf = Persistence.createEntityManagerFactory("xyz.schema", properties);
}
public static EntityManagerFactory getEntityManagerFactory() {
return emf;
}
public static EntityManager getEntityManager() {
EntityManager em = threadLocal.get();
if (em == null || !em.isOpen()) {
em = emf.createEntityManager();
entityManagers.put(em.toString(), em);
threadLocal.set(em);
}
return em;
}
public static <T> TypedQuery<T> createQuery(String qlString, Class<T> resultClass) {
getEntityManager().clear();
return getEntityManager().createQuery(qlString, resultClass);
}
public static Query createQuery(String qlString) {
return getEntityManager().createQuery(qlString);
}
public static void closeEntityManager() {
EntityManager em = threadLocal.get();
if (em != null) {
entityManagers.remove(em.toString());
em.close();
threadLocal.set(null);
}
}
public static void clearEntityManager() {
EntityManager em = threadLocal.get();
if (em != null) {
em.clear();
}
}
public static void closeEntityManagerFactory() {
emf.close();
emf = null;
}
public static void beginTransaction() {
getEntityManager().getTransaction().begin();
}
public static void rollback() {
getEntityManager().getTransaction().rollback();
}
public static void commit() {
getEntityManager().getTransaction().commit();
}
public static void detach(Object entity) {
Iterator it = entityManagers.values().iterator();
while (it.hasNext()) {
EntityManager em = (EntityManager) it.next();
if (em != null || em.isOpen())
em.detach(entity);
}
}
public static void evict(Object entity) {
Iterator it = entityManagers.values().iterator();
while (it.hasNext()) {
EntityManager em = (EntityManager) it.next();
if (em != null || em.isOpen())
em.getEntityManagerFactory().getCache().evict(entity.getClass(), ((EntityBase) entity).getId());
//em.getEntityManagerFactory().createEntityManager(SynchronizationType.)
}
}
public static void merge(Object entity) {
Iterator it = entityManagers.values().iterator();
while (it.hasNext()) {
EntityManager em = (EntityManager) it.next();
if (em != null || em.isOpen())
em.merge(entity);
}
}
public static void refresh(Object entity) {
Iterator it = entityManagers.values().iterator();
while (it.hasNext()) {
EntityManager em = (EntityManager) it.next();
if (em != null || em.isOpen())
em.refresh(entity);
}
}
public static boolean isActive() {
return getEntityManager().getTransaction().isActive();
}
public static boolean isOpen() {
return emf.isOpen();
}
public static void evictAll() {
emf.getCache().evictAll();
}
}
【问题讨论】：
标签： java multithreading jpa entitymanager