我的应用程序使用 JPA/EclipseLink 和 MySQL 作为数据库,遇到了并发问题。我在这里需要建议。
我有一个客户端,它在文件中生成字符串 block 。每个 block 可能包含几行到 100 或 200 行。生成数据后,客户端开始将一组 block 发送到服务器,其中每个 block 都会被相应地解析并存储在数据库中。服务器使用JPA/EclipseLink与MySQL通信。
当客户端以同步方式(通过 REST)发送一组 block 时,解析数据并将其存储在数据库中的一切都按预期进行。然而,当客户端并行发送数据时(在我的例子中是 15 个并发线程),一些数据(几行)仅在第一次存储在数据库中时被跳过。当相同的数据再次并行发送时(第二次、第三次等),它会按预期工作(存储期间不会跳过任何行)。
为了在并行发送的情况下进行测试,我将所有行存储在并发映射中。我没有在 map 中看到任何跳过的行,因此这纯粹是 JPA 和/或 MySQL 问题。我尝试在 JPA 中使用乐观/悲观锁,但它们没有帮助。
您能告诉我,如果有人遇到这种情况怎么办?
这是我的三个相关实体类:
@Entity
@Table(name = "PROJECT")
@NamedQueries({
@NamedQuery(name = "Project.findByKey", query = "SELECT p FROM Project p WHERE p.key = :P_KEY",
hints = {
@QueryHint(name = QueryHints.QUERY_RESULTS_CACHE, value = HintValues.TRUE),
@QueryHint(name = QueryHints.QUERY_RESULTS_CACHE_SIZE, value = "1000"),
@QueryHint(name = QueryHints.QUERY_RESULTS_CACHE_IGNORE_NULL, value = HintValues.TRUE),
@QueryHint(name = QueryHints.BIND_PARAMETERS, value = HintValues.TRUE)
}
)
})
@XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
@XmlType
@CascadeOnDelete
@OptimisticLocking
public class Project implements Serializable {
@Id
@TableGenerator(name = "p-table-gen", table = "SEQ", pkColumnName = "SEQ_NAME", valueColumnName = "SEQ_COUNT", pkColumnValue = "PROJECT_SEQ", allocationSize = 50)
@GeneratedValue(strategy = GenerationType.TABLE, generator = "p-table-gen")
private Long id;
@Column(name = "DURATION")
private Long duration;
@Column(name = "PROGRAM")
private String program;
@Embedded
@AttributeOverrides(
{
@AttributeOverride(name = "parentHash", column = @Column(name = "PARENT_CMD_HASH")),
@AttributeOverride(name = "hash", column = @Column(name = "CMD_HASH")),
@AttributeOverride(name = "pathHash", column = @Column(name = "PATH_HASH"))
}
)
ProjectKey key = new ProjectKey();
@BatchFetch(BatchFetchType.EXISTS)
@OneToMany(mappedBy="project", cascade = CascadeType.ALL, fetch = FetchType.LAZY, orphanRemoval = true )
@XmlTransient
@CascadeOnDelete
private List<Task> tasks = new Vector<Task>();
@BatchFetch(BatchFetchType.EXISTS)
@OneToOne(mappedBy="project", cascade = CascadeType.ALL, fetch = FetchType.LAZY, orphanRemoval = true )
@XmlTransient
@CascadeOnDelete
private Environment environment;
@Version
@Column(name = "VERSION")
private Long version;
}
@Entity
@Table(name = "TASK")
@NamedQueries({
@NamedQuery(name = "Task.findTaskByBuildAndPathName",
query = "SELECT t FROM Task t WHERE t.operation = 'W' AND t.key.pathName = :PATH_NAME ORDER BY t.startTime",
hints = {
@QueryHint(name = QueryHints.BIND_PARAMETERS, value = HintValues.TRUE)
}
)
})
@XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
@XmlType
@CascadeOnDelete
@OptimisticLocking
public class Task implements Serializable, Constants {
private static final long serialVersionUID = 1L;
@Id
@TableGenerator(name = "t-table-gen", table = "SEQ", pkColumnName = "SEQ_NAME", valueColumnName = "SEQ_COUNT", pkColumnValue = "TASK_SEQ", allocationSize = 200)
@GeneratedValue(strategy = GenerationType.TABLE, generator = "t-table-gen")
private Long id;
@Column(name = "HASH")
private String hash;
@Column(name = "OPERATION")
private String operation;
@Column(name = "PROCESS_ID")
private BigInteger processId;
@Embedded
@AttributeOverride(name = "nanoseconds", column = @Column(name = "START_TIME"))
private Moment startTime;
@Column(name = "THREAD_ID")
private BigInteger threadId;
@Column(name = "THROUGHPUT")
private Float throughput;
@Column(name = "PROCESSED_BYTES")
private Float processedBytes;
@BatchFetch(BatchFetchType.JOIN)
@ManyToOne(cascade = { CascadeType.MERGE, CascadeType.REMOVE}, fetch=FetchType.EAGER)
@JoinColumn(name = "PATH_STATE_ID", insertable = false, updatable = true)
@CascadeOnDelete
private TaskKey key;
@ManyToOne
@JoinColumn(name = "PROJECT_ID")
@XmlTransient
@CascadeOnDelete
private Project project;
@Version
@Column(name = "VERSION")
private Long version;
}
@Entity
@Index(name="IDX_TS_SIZE_TIME_INDEX", columnNames={"SIZE","TIME"})
@Table(name = "TASK_STATE")
@NamedQueries({
@NamedQuery(name = "TaskState.findByKey", query = "SELECT ts FROM TaskState ts WHERE ts.key = :TS_KEY",
hints = {
@QueryHint(name = QueryHints.QUERY_RESULTS_CACHE, value = HintValues.TRUE),
@QueryHint(name = QueryHints.QUERY_RESULTS_CACHE_SIZE, value = "2000"),
@QueryHint(name = QueryHints.QUERY_RESULTS_CACHE_IGNORE_NULL, value="true"),
@QueryHint(name = QueryHints.BIND_PARAMETERS, value = HintValues.TRUE)
}
)
})
@XmlRootElement
@XmlAccessorType(XmlAccessType.FIELD)
@XmlType
public class TaskState implements Serializable {
@Id
@TableGenerator(name = "ts-table-gen", table = "SEQ", pkColumnName = "SEQ_NAME", valueColumnName = "SEQ_COUNT", pkColumnValue = "TASK_STATE_SEQ", allocationSize = 50)
@GeneratedValue(strategy = GenerationType.TABLE, generator = "ts-table-gen")
private Long id;
@Lob
@Basic(fetch = FetchType.LAZY)
private byte[] artifact;
@Column(name = "MODE")
private String mode;
@Embedded
@AttributeOverrides({
@AttributeOverride(name = "nanoseconds", column = @Column(name = "TIME")),
@AttributeOverride(name = "size", column = @Column(name = "SIZE")),
@AttributeOverride(name = "pathName", column = @Column(name = "NAME")),
@AttributeOverride(name = "fileHash", column = @Column(name = "HASH"))
})
TaskStateKey key = new TaskStateKey();
@BatchFetch(BatchFetchType.EXISTS)
@OneToMany(mappedBy = "taskState", orphanRemoval = true, cascade = { CascadeType.REMOVE})
@XmlTransient
@CascadeOnDelete
private List<Task> tasks = new Vector<Task>();
}
更新
现在我想我找到了根本问题,但我还没有找到解决方案。
最初,我在数据库上创建一个项目,然后尝试查找相关的任务状态。最初没有任务状态,所以我一一创建。因此,这是有关 taskState 的操作的结果:
1. 尝试通过key查找taskState,如果有则返回,否则
2.持久/插入新的taskState(当有多个线程尝试执行插入操作时,我收到重复的条目错误,因此我只是重试插入操作直到成功。)并返回新创建的pathState。
每当我在当前的 pathState 插入操作中收到重复的条目错误时,我就会看到跳过的行,即使是在第二次或第三次我成功地保留了 pathState 时也是如此。我可以在数据库上启用 SERIALIZABLE 隔离级别,但这效率很低,在我的情况下我无法承受,并且我怀疑 SERIALIZABLE 隔离能否解决我的问题。
最佳答案
将TaskState的ID生成策略从TABLE更改为IDENTITY;这将允许 MySQL 自行设置插入行的 ID。
关于java - JPA/EclipseLink 与 MySQL 的并发问题,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/24692328/