什么是activeMQ
activeMQ是一种开源的,实现了JMS1.1规范的,面向消息(MOM)的中间件,为应用程序提供高效的、可扩展的、稳定的和安全的企业级消息通信。
ActiveMQ的作用以及原理
Activemq 的作用就是系统之间进行通信。 当然可以使用其他方式进行系统间通信, 如果使用 Activemq 的话可以对系统之间的调用进行解耦, 实现系统间的异步通信。 原理就是生产者生产消息, 把消息发送给activemq。 Activemq 接收到消息, 然后查看有多少个消费者, 然后把消息转发给消费者, 此过程中生产者无需参与。 消费者接收到消息后做相应的处理和生产者没有任何关系
activemq的几种通信方式
(1)publish(发布)-subscribe(订阅)(发布-订阅方式)
发布/订阅方式用于多接收客户端的方式.作为发布订阅的方式,可能存在多个接收客户端,并且接收端客户端与发送客户端存在时间上的依赖。一个接收端只能接收他创建以后发送客户端发送的信息。作为subscriber ,在接收消息时有两种方法,destination的receive方法,和实现message listener 接口的onMessage 方法。
(2)p2p(point-to-point)(点对点)
p2p的过程则理解起来比较简单。它好比是两个人打电话,这两个人是独享这一条通信链路的。一方发送消息,另外一方接收,就这么简单。
ActiveMQ服务器宕机怎么办?
这得从ActiveMQ的储存机制说起。在通常的情况下,非持久化消息是存储在内存中的,持久化消息是存储在文件中的,它们的最大限制在配置文件的<systemUsage>节点中配置。但是,在非持久化消息堆积到一定程度,内存告急的时候,ActiveMQ会将内存中的非持久化消息写入临时文件中,以腾出内存。虽然都保存到了文件里,但它和持久化消息的区别是,重启后持久化消息会从文件中恢复,非持久化的临时文件会直接删除。
解决方案:尽量不要用非持久化消息,非要用的话,将临时文件限制尽可能的调大。
丢消息怎么办?
这得从java的java.net.SocketException异常说起。简单点说就是当网络发送方发送一堆数据,然后调用close关闭连接之后。这些发送的数据都在接收者的缓存里,接收者如果调用read方法仍旧能从缓存中读取这些数据,尽管对方已经关闭了连接。但是当接收者尝试发送数据时,由于此时连接已关闭,所以会发生异常,这个很好理解。不过需要注意的是,当发生SocketException后,原本缓存区中数据也作废了,此时接收者再次调用read方法去读取缓存中的数据,就会报Software caused connection abort: recv failed错误。
通过抓包得知,ActiveMQ会每隔10秒发送一个心跳包,这个心跳包是服务器发送给客户端的,用来判断客户端死没死。如果你看过上面第一条,就会知道非持久化消息堆积到一定程度会写到文件里,这个写的过程会阻塞所有动作,而且会持续20到30秒,并且随着内存的增大而增大。当客户端发完消息调用connection.close()时,会期待服务器对于关闭连接的回答,如果超过15秒没回答就直接调用socket层的close关闭tcp连接了。这时客户端发出的消息其实还在服务器的缓存里等待处理,不过由于服务器心跳包的设置,导致发生了java.net.SocketException异常,把缓存里的数据作废了,没处理的消息全部丢失。
解决方案:用持久化消息,或者非持久化消息及时处理不要堆积,或者启动事务,启动事务后,commit()方法会负责任的等待服务器的返回,也就不会关闭连接导致消息丢失了。
持久化消息非常慢
默认的情况下,非持久化的消息是异步发送的,持久化的消息是同步发送的,遇到慢一点的硬盘,发送消息的速度是无法忍受的。但是在开启事务的情况下,消息都是异步发送的,效率会有2个数量级的提升。所以在发送持久化消息时,请务必开启事务模式。其实发送非持久化消息时也建议开启事务,因为根本不会影响性能。
消息的不均匀消费
有时在发送一些消息之后,开启2个消费者去处理消息。会发现一个消费者处理了所有的消息,另一个消费者根本没收到消息。原因在于ActiveMQ的prefetch机制。当消费者去获取消息时,不会一条一条去获取,而是一次性获取一批,默认是1000条。这些预获取的消息,在还没确认消费之前,在管理控制台还是可以看见这些消息的,但是不会再分配给其他消费者,此时这些消息的状态应该算作“已分配未消费”,如果消息最后被消费,则会在服务器端被删除,如果消费者崩溃,则这些消息会被重新分配给新的消费者。但是如果消费者既不消费确认,又不崩溃,那这些消息就永远躺在消费者的缓存区里无法处理。更通常的情况是,消费这些消息非常耗时,你开了10个消费者去处理,结果发现只有一台机器吭哧吭哧处理,另外9台啥事不干。
解决方案:将prefetch设为1,每次处理1条消息,处理完再去取,这样也慢不了多少。
死信队列
如果你想在消息处理失败后,不被服务器删除,还能被其他消费者处理或重试,可以关闭AUTO_ACKNOWLEDGE,将ack交由程序自己处理。那如果使用了AUTO_ACKNOWLEDGE,消息是什么时候被确认的,还有没有阻止消息确认的方法?有!
消费消息有2种方法,一种是调用consumer.receive()方法,该方法将阻塞直到获得并返回一条消息。这种情况下,消息返回给方法调用者之后就自动被确认了。另一种方法是采用listener回调函数,在有消息到达时,会调用listener接口的onMessage方法。在这种情况下,在onMessage方法执行完毕后,消息才会被确认,此时只要在方法中抛出异常,该消息就不会被确认。那么问题来了,如果一条消息不能被处理,会被退回服务器重新分配,如果只有一个消费者,该消息又会重新被获取,重新抛异常。就算有多个消费者,往往在一个服务器上不能处理的消息,在另外的服务器上依然不能被处理。难道就这么退回--获取--报错死循环了吗?
在重试6次后,ActiveMQ认为这条消息是“有毒”的,将会把消息丢到死信队列里。如果你的消息不见了,去ActiveMQ.DLQ里找找,说不定就躺在那里。
java消息服务
不同系统之间的信息交换,是我们开发中比较常见的场景,比如系统A要把数据发送给系统B,这个问题我们应该如何去处理? 1999年,原来的SUN公司领衔提出了一种面向消息的中间件服务--JMS规范(标准);常用的几种信息交互技术(httpClient、hessian、dubbo、jms、webservice 五种).
JMS概述
JMS即Java消息服务(Java Message Service的简称),是Java EE 的标准/规范之一。这种规范(标准)指出:消息的发送应该是异步的、非阻塞的。也就是说消息的发送者发送完消息后就直接返回了,不需要等待接收者返回后才能返回,发送者和接收者可以说是互不影响。所以这种规范(标准)能够减轻或消除系统瓶颈,实现系统之间去除耦合,提高系统的整体可伸缩性和灵活性。JMS只是Java EE中定义的一组标准API,它自身并不是一个消息服务系统,它是消息传送服务的一个抽象,也就是说它定义了消息传送的接口而并没有具体实现。
ActiveMQ概述:
我们知道JMS只是消息服务的一组规范和接口,并没有具体的实现,而ActiveMQ就是JMS规范的具体实现;它是Apache下的一个项目,采用Java语言开发;是一款非常流行的开源消息服务器.
ActiveMQ与JMS关系
我们知道,JMS只是定义了一组有关消息传送的规范和标准,并没有真正实现,也就说JMS只是定义了一组接口而已;就像JDBC抽象了关系数据库访问、JPA抽象了对象与关系数据库映射、JNDI抽象了命名目录服务访问一样,JMS具体的实现由不同的消息中间件厂商提供,比如Apache ActiveMQ就是JMS规范的具体实现,Apache ActiveMQ才是一个消息服务系统,而JMS不是。
基本要素
1、生产者producer ; 2、消费者consumer ; 3、消息服务broker
JMS两种消息传送模式
点对点( Point-to-Point):专门用于使用队列Queue传送消息;基于队列Queue的点对点消息只能被一个消费者消费,如多个消费者都注册到同一个消息队列上,当生产者发送一条消息后,而只有其中一个消费者会接收到该消息,而不是所有消费者都能接收到该消息。
发布/订阅(Publish/Subscribe):专门用于使用主题Topic传送消息。基于主题的发布与订阅消息能被多个消费者消费,生产者发送的消息,所有订阅了该topic的消费者都能接收到。
JMS API 概览
JMS API可以分为3个主要部分:
1、公共API:可用于向一个队列或主题发送消息或从其中接收消息;
2、点对点API:专门用于使用队列Queue传送消息;
3、发布/订阅API:专门用于使用主题Topic传送消息。
JMS公共API
在JMS公共API内部,和发送与接收消息有关的JMS API接口主要是:ConnectionFactory / Connection / Session / Message / Destination / MessageProducer / MessageConsumer . 它们的关系是:一旦有了ConnectionFactory,就可以创建Connection,一旦有了Connection,就可以创建Session,而一旦有了Session,就可以创建 Message 、MessageProducer 和 MessageConsumer 。
JMS点对点API
点对点(p2p)消息传送模型API是指JMS API之内基于队列(Queue)的接口:QueueConnectionFactory / QueueConnection / QueueSession / Message / Queue / QueueSender / QueueReceiver .
从接口的命名可以看出,大多数接口名称仅仅是在公共API接口
名称之前添加Queue一词。一般来说,使用点对点消息传送模型的应用程序将使用基于队列的API,而不使用公共API 。
JMS发布/订阅API发布/订阅消息传送模型API是指JMS
API之内基于主题(Topic)的接口:TopicConnectionFactory / TopicConnection /
TopicSession / Message / Topic / TopicPublisher / TopicSubscriber . 由于基于主题(Topic)的JMS API类似于基于队列(Queue)
的API,因此在大多数情况下,Queue这个词会由Topic取代。
ActiveMQ点对点发送与接收消息示例
简单示例
发送
package com.kinglong.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
/**
* 消息发送者
*
*/
public class Sender {
/**消息服务器的连接地址**/
public static final String BROKER_URL = "tcp://192.168.174.129:61616";
public static void main(String[] args) {
Sender sender = new Sender();
sender.sendMessage("Hello ActiveMQ.");
}
/**
* 发送消息
*
* @param msg
*/
public void sendMessage (String msg) {
Connection connection = null;
Session session = null;
MessageProducer messageProducer = null;
try {
//1.创建一个连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
//2.创建一个连接
connection = connectionFactory.createConnection();
//3.创建一个Session
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//4.创建消息,此处创建了一个文本消息
Message message = session.createTextMessage(msg);
//5.创建一个目的地
Destination destination = session.createQueue("myQueue");
//6.创建一个消息的生产者(发送者)
messageProducer = session.createProducer(destination);
//7.发送消息
messageProducer.send(message);
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
//关闭连接释放资源
if (null != messageProducer) {
messageProducer.close();
}
if (null != session) {
session.close();
}
if (null != connection) {
connection.close();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
接收
package com.kinglong.activemq.queue;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Receiver {
/**消息服务器的连接地址**/
public static final String BROKER_URL = "tcp://192.168.174.129:61616";
public static void main(String[] args) {
Receiver receiver = new Receiver();
receiver.receiveMessage();
}
/**
* 接收消息
*
*/
public void receiveMessage () {
Connection connection = null;
Session session = null;
MessageConsumer messageConsumer = null;
try {
//1.创建一个连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
//2.创建一个连接
connection = connectionFactory.createConnection();
//3.创建一个Session
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//4.创建一个目的地
Destination destination = session.createQueue("myQueue");
//5.创建一个消息的消费者(接收者)
messageConsumer = session.createConsumer(destination);
//接收消息之前,需要把连接启动一下
connection.start();
//6.接收消息
Message message = messageConsumer.receive();
//判断消息的类型
if (message instanceof TextMessage) { //判断是否是文本消息
String text = ((TextMessage) message).getText();
System.out.println("接收到的消息内容是:" + text);
}
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
//关闭连接释放资源
if (null != messageConsumer) {
messageConsumer.close();
}
if (null != session) {
session.close();
}
if (null != connection) {
connection.close();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
总结
1. session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
// 其中:Boolean.FALSE表示本次会话不开启事务管理,假如需要开启事务管理,将其改为Boolean.TRUE即可
//同时需要在发送消息后添加session.commit(),否则,消息是不会被提交的.
//Session.AUTO_ACKNOWLEDGE表示消息确认机制
AUTO_ACKNOWLEDGE:自动确认
CLIENT_ACKNOWLEDGE:客户端确认
SESSION_TRANSACTED:事务确认,如果使用事务推荐使用该确认机制
AUTO_ACKNOWLEDGE:懒散式确认,消息偶尔不会被确认,也就是消息可能会被重复发送.但发生的概率很小
2. connection.start();
//在消息接收端,接受消息前需要加入这段代码,开启连接,否则一样无法获取消息.
3. Destination destination = session.createQueue("myQueue");
//创建目的地时,如果做测试收不到信息,可以将目的地名称修改一下,我用的是IDEA,不清楚为何,
//有时候收不到信息,修改一下就好了,猜测可能是缓存的原因吧
发布与订阅的topic方式实际与点对点的queue方式,代码通用很多,只是在创建目的地Destination时候创建为
Destination destination = session.createTopic("myTopic
Queue与Topic比较
比较项 | 队列 | 主题 |
---|---|---|
概要说明 | point-to-ponit点对点小欻 | publish/subscribe发布订阅消息 |
消息完整性保障 | 队列保证每条数据都能被receiver接收 | 不保证publisher发布的每条数据,subscriber都能接收到 |
消息是否会丢失 | sender发送消息到目标队列,receiver可以同步或者异步接收到这个队列上的消息,队列上的消息如果暂时没有被receiver接收到,也不会丢失 | publisher发布消息到目标主题时,只有正在监听该主题的subscriber能够接收到消息,如果目标topic没有subscriber在监听,该主题上的消息就会丢失。 |
消息传送策略 | 一对一的消息发布接受策略,一个sender发送的消息,只能有一个receiver接收,receiver接收到后,通知mq服务器已接收,mq服务器对队列里的消息采取删除或其他操作。 | 一对多的消息发布接收策略,监听同一个主题地址的多个subscriber都能接收到publisher发送的消息,subscriber接收完通知mq服务器。 |
拉模式与推模式
a.点对点消息,如果没有消费者在监听队列,消息将保留在队列中,直至消息消费者连接到队列为止。这种消息传递模型是
传统意义上的懒模型或轮询模型。在此模型中,消息不是自动推动给消息消费者的,而是要由消息消费者从队列中请求获得(拉模式)。
b.pub/sub消息传递模型基本上是一个推模型。在该模型中,消息会自动广播,消息消费者无须通过主动请求或轮询主题的方法来获得新的消息。
ActiveMQ消息类型
1、TextMessage 文本消息:携带一个java.lang.String作为有效数据(负载)的消息,可用于字符串类型的信息交换;
2、ObjectMessage 对象消息:携带一个可以序列化的Java对象作为有效负载的消息,可用于Java对象类型的信息交换;
3、MapMessage 映射消息:携带一组键值对的数据作为有效负载的消息,有效数据值必须是Java原始数据类型(或者它们的包装类)及String。即:byte , short , int , long , float , double , char , boolean , String
4、BytesMessage 字节消息 :携带一组原始数据类型的字节流作为有效负载的消息;
5、StreamMessage 流消息:携带一个原始数据类型流作为有效负载的消息,它保持了写入流时的数据类型,写入什么类型,则读取也需要是相同的类型;
ActiveMQ事务消息和非事务消息
消息分为事务消息和非事务消息
1、事务消息:创建会话Session使用transacted=true
connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);
2、非事务消息:创建会话Session使用transacted=false
connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
事务消息必须在发送和接收完消息后显式地调用session.commit();
事务性消息,不管设置何种消息确认模式,都会自动被确认;与设置的确认机制无关,但官方推荐事务性消息使用事务确认机制.
ActiveMQ消息确认机制
消息只有在被确认之后,才认为已经被成功消费,然后消息才会从队列或主题中删除。消息的成功消费通常包含三个阶段:
(1)、客户接收消息;(2)、客户处理消息;(3)、消息被确认;
确认机制前面提过一下,共有四种:
(1)、Session.AUTO_ACKNOWLEDGE;客户(消费者)成功从receive方法返回时,或者从MessageListener.onMessage方法成功返回时,会话自动确认消息,然后自动删除消息.
(2)、Session.CLIENT_ACKNOWLEDGE;客户通过显式调用消息的acknowledge方法确认消息,。 即在接收端调用message.acknowledge();方法,否则,消息是不会被删除的.
(3)、Session. DUPS_OK_ACKNOWLEDGE ;不是必须确认,是一种“懒散的”消息确认,消息可能会重复发送,在第二次重新传送消息时,消息头的JMSRedelivered会被置为true标识当前消息已经传送过一次,客户端需要进行消息的重复处理控制。
(4)、 Session.SESSION_TRANSACTED;事务提交并确认。
ActiveMQ持久化消息与非持久化消息
messageProducer.setDeliveryMode(DeliveryMode. NON_PERSISTENT);//不持久化
messageProducer.setDeliveryMode(DeliveryMode.
PERSISTENT);//持久化的,当然activemq发送消息默认都是持久化的
说明:
设置完后,如果为持久化,那么消息在没有被消费前都会被写入本地磁盘kahadb文件中保存起来,即使服务器宕机,也不会影响
消息.如果是非持久化的,那么,服务一旦宕机之类的情况发生,消息即会被删除.
ActiveMQ默认是持久化的.
ActiveMQ消息过滤
ActiveMQ提供了一种机制,可根据消息选择器中的标准来执行消息过滤,只接收符合过滤标准的消息;
生产者可在消息中放入特有的标志,而消费者使用基于这些特定的标志来接收消息;
1、发送消息放入特殊标志:message . setString Property ( name , value ) ;
2、接收消息使用基于特殊标志的消息选择器:
MessageConsumer createConsumer(Destination destination, String messageSelector);
注:消息选择器是一个字符串,语法与数据库的SQL相似,相当于SQL语句where条件后面的内容;
具体代码如下:
发送端代码:
package com.bjpowernode.activemq.selector;
import org.apache.activemq.Active
MQConnectionFactory;
import javax.jms.*;
/**
* 消息发送者
*
*/
public class Sender {
/**消息服务器的连接地址**/
public static final String BROKER_URL = "tcp://192.168.174.129:61616";
public static void main(String[] args) {
Sender sender = new Sender();
sender.sendMessage("Hello ActiveMQ.");
}
/**
* 发送消息
*
* @param msg
*/
public void sendMessage (String msg) {
Connection connection = null;
Session session = null;
MessageProducer messageProducer = null;
try {
//1.创建一个连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
//2.创建一个连接
connection = connectionFactory.
createConnection();
//3.创建一个Session
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//5.创建一个目的地
Destination destination = session.create
Queue("myQueue");
//6.创建一个消息的生产者(发送者)
messageProducer = session.createProducer
(destination);
//设置发送的消息是否需要持久化
messageProducer.setDeliveryMode(Delivery
Mode.NON_PERSISTENT);//这里使用不持久化
//创建一个循环,测试消息标识的使用
for (int i=0; i<20; i++) {
//4.创建消息,此处创建了一个文本消息
Message message = session.createText
Message(msg+i);
//将消息设置一个特有的标识
message.setIntProperty("id", i);
//7.发送消息
messageProducer.send(message);
}
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
//关闭连接释放资源
if (null != messageProducer) {
messageProducer.close();
}
if (null != session) {
session.close();
}
if (null != connection) {
connection.close();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
接收端代码:
package com.bjpowernode.activemq.selector;
import org.apache.activemq.ActiveMQ
ConnectionFactory;
import javax.jms.*;
public class Receiver {
/**消息服务器的连接地址**/
public static final String BROKER_URL = "tcp://192.168.174.129:61616";
public static void main(String[] args) {
Receiver receiver = new Receiver();
receiver.receiveMessage();
}
/**
* 接收消息
*
*/
public void receiveMessage () {
Connection connection = null;
Session session = null;
MessageConsumer messageConsumer = null;
try {
//1.创建一个连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
//2.创建一个连接
connection = connectionFactory.
createConnection();
//3.创建一个Session
session = connection.createSession
(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//4.创建一个目的地
Destination destination = session.createQueue("myQueue");
//5.创建一个消息的消费者(接收者),
selector即为消息选择器,通过选择需要的标识,过滤消息接受id为10-15之 //间的消息
String selector = "id >=10 and id<=15";
messageConsumer = session.createConsumer
(destination, selector);
//接收消息之前,需要把连接启动一下
connection.start();
while (true) {
//6.接收消息 同步接收
Message message = messageConsumer.
receive();
//判断消息的类型
if (message instanceof TextMessage)
{ //判断是否是文本消息
String text = ((TextMessage) message).getText();
System.out.println("接收到的消息内容是:" + text);
}
}
} catch (JMSException e) {
e.printStackTrace();
} finally {
try {
//关闭连接释放资源
if (null != messageConsumer) {
messageConsumer.close();
}
if (null != session) {
session.close();
}
if (null != connection) {
connection.close();
}
} catch (JMSException e) {
e.printStackTrace();
}
}
}
}
ActiveMQ消息接收方式
同步接收:receive()方法接收消息叫同步接收,就是之前的Demo代码使用的接收方式.在不使用循环方法时接收端代码执行
一次即结束.
异步接收:使用监听器接收消息,这种接收方式叫异步接收,接收端会一直处于监听状态,只要有消息产生,即会接收消息.
下面是异步接收代码:
package com.bjpowernode.activemq.listener;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;
public class Receiver {
/**消息服务器的连接地址**/
public static final String BROKER_URL = "tcp://192.168.174.129:61616";
public static void main(String[] args) {
Receiver receiver = new Receiver();
receiver.receiveMessage();
}
/**
* 接收消息
*
*/
public void receiveMessage () {
Connection connection = null;
Session session = null;
MessageConsumer messageConsumer = null;
try {
//1.创建一个连接工厂
ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(BROKER_URL);
//2.创建一个连接
connection = connectionFactory.createConnection();
//3.创建一个Session
session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
//4.创建一个目的地
Destination destination = session.createQueue("myQueue");
//5.创建一个消息的消费者(接收者)
messageConsumer = session.createConsumer(destination);
//接收消息之前,需要把连接启动一下
connection.start();
//6.接收消息 同步接收
//Message message = messageConsumer.receive();
//异步接收,使用监听器接收消息
messageConsumer.setMessageListener(new MessageListener(){
public void onMessage(Message message) {
//判断消息的类型
if (message instanceof TextMessage) { //判断是否是文本消息
String text = null;
try {
text = ((TextMessage) message).getText();
} catch (JMSException e) {
e.printStackTrace();
}
System.out.println("接收到的消息内容是:" + text);
}
}
});
} catch (JMSException e) {
e.printStackTrace();
} finally {
/*try {
//关闭连接释放资源
if (null != messageConsumer) {
messageConsumer.close();
}
if (null != session) {
session.close();
}
if (null != connection) {
connection.close();
}
} catch (JMSException e) {
e.printStackTrace();
}*/
}
}
}