![cover](https://img-blog.csdnimg.cn/img_convert/1f20816e15694cf3b56262c1f87e2a14.png)
Rabbitmq消息顺序的问题以及解决方案
场景1:一个queue,多个consumer一个queue,有多个consumer去消费,这样就会造成顺序的错误,consumer从MQ里面读取数据是有序的,但是每个consumer的执行时间是不固定的,无法保证先读到消息的consumer一定先完成操作,这样就会出现消息并没有按照顺序执行,造成数据顺序错误。人话就是:我确实保证了消息是按按顺序接的。但是由于每一个消息执行的是时间不一样。如果我前面
·
1.1消息顺序的场景
场景1:一个queue,多个consumer
一个queue,有多个consumer去消费,这样就会造成顺序的错误,consumer从MQ里面读取数据是有序的,但是每个consumer的执行时间是不固定的,无法保证先读到消息的consumer一定先完成操作,这样就会出现消息并没有按照顺序执行,造成数据顺序错误。,
人话就是:我确实保证了消息是按按顺序接的。但是由于每一个消息执行的是时间不一样。如果我前面的执行消息比较长,会导致我后面的操作比前面的操作更早执行。就出现了顺序错误
consumer多线程消费
一个queue对应一个consumer,但是consumer里面进行了多线程消费,这样也会造成消息消费顺序错误。
人话:就是线程的执行是强制式的,你永远不知道下一个执行的会是哪个指令。
2.解决方案:
单消费者、单线程消费就行。
如果你非要多线程,你可以参考下,多线程输出a,b,c
/**
* @author XDarker
* 2018-5-17
*/
public class Main {
public static void main(String[] args) throws InterruptedException {
int num = 1;//当前正在执行线程的标记
ABCPrint print = new ABCPrint(num);
Thread threadA = new Thread(new RunnableA(print));
Thread threadB = new Thread(new RunnableB(print));
Thread threadC = new Thread(new RunnableC(print));
threadA.start();
Thread.sleep(500);
threadB.start();
Thread.sleep(500);
threadC.start();
}
}
class RunnableA implements Runnable{
private ABCPrint print;
public RunnableA(ABCPrint print) {
super();
this.print = print;
}
@Override
public void run() {
print.PrintA();
}
}
class RunnableB implements Runnable{
private ABCPrint print;
public RunnableB(ABCPrint print) {
super();
this.print = print;
}
@Override
public void run() {
print.PrintB();
}
}
class RunnableC implements Runnable{
private ABCPrint print;
public RunnableC(ABCPrint print) {
super();
this.print = print;
}
@Override
public void run() {
print.PrintC();
}
}
class ABCPrint {
private int num;//当前正在执行线程的标记
public ABCPrint(int num) {
super();
this.num = num;
}
public void PrintA(){
for (int j = 0; j < 2; j++)//表示 循环打印2轮
synchronized(this){
while(num != 1){
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
for (int i = 0; i < 3; i++) {//表示 打印3次
System.out.println("A");
}
//打印A线程执行完 ,通知打印B线程
num = 2;
this.notifyAll();
}
}
public void PrintB(){
for (int j = 0; j < 2; j++)//表示 循环打印2轮
synchronized(this){
while(num != 2){
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
for (int i = 0; i < 2; i++) {//表示 打印2次
System.out.println("B");
}
//打印B线程执行完 ,通知打印C线程
num = 3;
this.notifyAll();
}
}
public void PrintC(){
for (int j = 0; j < 2; j++)//表示 循环打印2轮
synchronized(this){
while(num != 3){
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println("C");
//打印C线程执行完 ,通知打印A线程
num = 1;
this.notifyAll();
}
}
}
更多推荐
所有评论(0)