diff --git a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java
index 70723daeb..8722e68da 100644
--- a/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java
+++ b/spring-batch-core/src/main/java/org/springframework/batch/core/configuration/xml/StepParser.java
@@ -271,14 +271,30 @@ public class StepParser {
 	        	DomUtils.getChildElementsByTagName(listenersElement, "listener");
 			if (listenerElements != null) {
 				for (Element listenerElement : listenerElements) {
+					String id = listenerElement.getAttribute("id");
 					String listenerRef = listenerElement.getAttribute("ref");
-						if (StringUtils.hasText(listenerRef)) {
-							listenerRefs.add(listenerRef);
-					        RuntimeBeanReference bean = new RuntimeBeanReference(listenerRef);
-							if (bean != null) {
-								listenerBeans.add(bean);
-							}
+					String className = listenerElement.getAttribute("class");
+					if ((StringUtils.hasText(id) || StringUtils.hasText(className)) 
+							&& StringUtils.hasText(listenerRef)) {
+						throw new BeanCreationException("Both 'id' or 'ref' plus 'class' specified; use 'class' with an optional 'id' or just 'ref' for <" + listenerElement.getTagName() + "> element with" + (StringUtils.hasText(id) ? " id=\"" + id + "\"" : "" ) + (StringUtils.hasText(className) ? " class=\"" + className + "\"" : "") + (StringUtils.hasText(listenerRef) ? " ref=\"" + listenerRef + "\"" : ""));
+					}
+					if (StringUtils.hasText(listenerRef)) {
+						listenerRefs.add(listenerRef);
+				        BeanReference bean = new RuntimeBeanReference(listenerRef);
+						listenerBeans.add(bean);
+					}
+					else if (StringUtils.hasText(className)) {
+						RootBeanDefinition beanDef = new RootBeanDefinition(className, null, null);
+						if (!StringUtils.hasText(id)) {
+							id = parserContext.getReaderContext().generateBeanName(beanDef);
 						}
+						parserContext.getRegistry().registerBeanDefinition(id, beanDef);
+				        BeanReference bean = new RuntimeBeanReference(id);
+						listenerBeans.add(bean);
+					}
+					else {
+						throw new BeanCreationException("Neither 'ref' or 'class' specified for <" + listenerElement.getTagName() + "> element");
+					}
 				}
 			}
 	        ManagedList arguments = new ManagedList();
diff --git a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd
index 6d665fea0..a1dd9da03 100644
--- a/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd
+++ b/spring-batch-core/src/main/resources/org/springframework/batch/core/configuration/xml/spring-batch-2.0.xsd
@@ -225,8 +225,9 @@
 						</xsd:annotation>
 						<xsd:complexType>
 							<xsd:sequence>
-								<xsd:element name="listener">
+								<xsd:element name="listener" minOccurs="1" maxOccurs="unbounded">
 									<xsd:complexType>
+										<xsd:attribute name="id" type="xsd:ID"/>
 										<xsd:attribute name="ref" type="xsd:string"/>
 										<xsd:attribute name="class" type="xsd:string"/>
 									</xsd:complexType>
diff --git a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests.java b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests.java
index 79627acf1..6dcc71787 100644
--- a/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests.java
+++ b/spring-batch-core/src/test/java/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests.java
@@ -29,6 +29,7 @@ import org.springframework.batch.core.JobParameters;
 import org.springframework.batch.core.repository.JobRepository;
 import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -51,6 +52,7 @@ public class StepWithChunkOrientedJobParserTests {
 	private TestReader reader;
 	
 	@Autowired
+	@Qualifier("listener")
 	private TestListener listener;
 	
 	@Autowired
diff --git a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests-context.xml b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests-context.xml
index 2fa0ba9dd..662599af5 100644
--- a/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests-context.xml
+++ b/spring-batch-core/src/test/resources/org/springframework/batch/core/configuration/xml/StepWithChunkOrientedJobParserTests-context.xml
@@ -10,6 +10,7 @@
 		<step name="step1">
 			<chunk-oriented reader="reader" processor="processor" writer="writer">
 				<listeners>
+					<listener class="org.springframework.batch.core.configuration.xml.TestListener"/>
 					<listener ref="listener"/>
 				</listeners>
 			</chunk-oriented>